From 6e34846c30a1020ff1605a1f41d8da9ed2a21a63 Mon Sep 17 00:00:00 2001 From: Tanner Stirrat Date: Fri, 2 Oct 2026 10:29:12 -0600 Subject: [PATCH] feat: add a rendezvous hash implementation --- README.md | 13 +++ algorithm_test.go | 153 +++++++++++++++++++++++++++ balancer.go | 71 +++++++++++-- rendezvous/compare_test.go | 191 +++++++++++++++++++++++++++++++++ rendezvous/rendezvous.go | 193 ++++++++++++++++++++++++++++++++++ rendezvous/rendezvous_test.go | 163 ++++++++++++++++++++++++++++ ringview.go | 8 +- 7 files changed, 785 insertions(+), 7 deletions(-) create mode 100644 algorithm_test.go create mode 100644 rendezvous/compare_test.go create mode 100644 rendezvous/rendezvous.go create mode 100644 rendezvous/rendezvous_test.go diff --git a/README.md b/README.md index 6889e41..5ff23d9 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,19 @@ balancer.Register(consistent.NewBuilder(xxhash.Sum64)) grpc.Dial(addr, grpc.WithDefaultServiceConfig(consistent.DefaultServiceConfigJSON)) ``` +### Choosing an algorithm + +By default, backends are placed on a consistent hashring with `ReplicationFactor` virtual nodes each. +Setting `Algorithm` to `consistent.AlgorithmRendezvous` uses rendezvous (highest random weight) hashing instead: + +```go +cfg := &consistent.BalancerConfig{Spread: 1, Algorithm: consistent.AlgorithmRendezvous} +grpc.Dial(addr, grpc.WithDefaultServiceConfig(cfg.MustServiceConfigJSON())) +``` + +Rendezvous hashing spreads keys more evenly and makes membership changes far cheaper, at the cost of lookups that scale linearly with the number of backends. +See `go test ./rendezvous -run TestCompareDistribution -v` and `go test ./rendezvous -run '^$' -bench .` for a comparison. + ## Acknowledgements This project is a community effort fueled by contributions from both organizations and individuals. diff --git a/algorithm_test.go b/algorithm_test.go new file mode 100644 index 0000000..97f1695 --- /dev/null +++ b/algorithm_test.go @@ -0,0 +1,153 @@ +package consistent + +import ( + "context" + "fmt" + "testing" + + "github.com/cespare/xxhash/v2" + "github.com/stretchr/testify/require" + "google.golang.org/grpc/balancer" + "google.golang.org/grpc/connectivity" + "google.golang.org/grpc/resolver" + + "github.com/authzed/consistent/hashring" + "github.com/authzed/consistent/rendezvous" +) + +func TestParseConfigAlgorithm(t *testing.T) { + bld := NewBuilder(xxhash.Sum64) + + cfg, err := bld.ParseConfig([]byte(`{"algorithm": "rendezvous"}`)) + require.NoError(t, err) + require.Equal(t, &BalancerConfig{ + ReplicationFactor: DefaultReplicationFactor, + Spread: DefaultSpread, + Algorithm: AlgorithmRendezvous, + }, cfg) + + cfg, err = bld.ParseConfig([]byte(`{"algorithm": "hashring"}`)) + require.NoError(t, err) + require.Equal(t, AlgorithmHashring, cfg.(*BalancerConfig).Algorithm) + + _, err = bld.ParseConfig([]byte(`{"algorithm": "maglev"}`)) + require.ErrorContains(t, err, `unknown algorithm "maglev"`) +} + +func TestServiceConfigJSONAlgorithm(t *testing.T) { + got, err := (&BalancerConfig{Spread: 1, Algorithm: AlgorithmRendezvous}).ServiceConfigJSON() + require.NoError(t, err) + require.Equal(t, `{"loadBalancingConfig":[{"consistent-hashring":{"spread":1,"algorithm":"rendezvous"}}]}`, got) +} + +// rendezvousBalancer builds a balancer that uses rendezvous hashing and +// moves every SubConn for addrs to READY. +func rendezvousBalancer(t *testing.T, addrs ...resolver.Address) *ringBalancer { + t.Helper() + b, _ := readyBalancer(t, addrs...) + require.NoError(t, b.UpdateClientConnState(balancer.ClientConnState{ + ResolverState: resolver.State{Addresses: addrs}, + BalancerConfig: &BalancerConfig{ReplicationFactor: 100, Spread: 1, Algorithm: AlgorithmRendezvous}, + })) + require.IsType(t, &rendezvous.Set{}, b.hashring) + return b +} + +// Switching algorithms rebuilds the member set. The new set must contain +// every member except those in TRANSIENT_FAILURE, and switching back must +// restore a hashring. +func TestAlgorithmChangeRebuildsMemberSet(t *testing.T) { + addrs := []resolver.Address{ + {ServerName: "t", Addr: "1"}, + {ServerName: "t", Addr: "2"}, + {ServerName: "t", Addr: "3"}, + } + b, _ := readyBalancer(t, addrs...) + require.IsType(t, &hashring.Ring{}, b.hashring) + + sci2, _ := b.subConns.Get(addrs[1]) + b.UpdateSubConnState(sci2.(balancer.SubConn), balancer.SubConnState{ + ConnectivityState: connectivity.TransientFailure, + ConnectionError: fmt.Errorf("refused"), + }) + + require.NoError(t, b.UpdateClientConnState(balancer.ClientConnState{ + ResolverState: resolver.State{Addresses: addrs}, + BalancerConfig: &BalancerConfig{ReplicationFactor: 100, Spread: 1, Algorithm: AlgorithmRendezvous}, + })) + require.IsType(t, &rendezvous.Set{}, b.hashring) + require.ElementsMatch(t, []string{"t1", "t3"}, ringKeys(b)) + + require.NoError(t, b.UpdateClientConnState(balancer.ClientConnState{ + ResolverState: resolver.State{Addresses: addrs}, + BalancerConfig: &BalancerConfig{ReplicationFactor: 100, Spread: 1}, + })) + require.IsType(t, &hashring.Ring{}, b.hashring) + require.ElementsMatch(t, []string{"t1", "t3"}, ringKeys(b)) +} + +// Rendezvous hashing has no virtual nodes, so a ReplicationFactor change must +// not rebuild the member set and move keys. +func TestRendezvousIgnoresReplicationFactor(t *testing.T) { + addrs := []resolver.Address{{ServerName: "t", Addr: "1"}, {ServerName: "t", Addr: "2"}} + b := rendezvousBalancer(t, addrs...) + before := b.hashring + + require.NoError(t, b.UpdateClientConnState(balancer.ClientConnState{ + ResolverState: resolver.State{Addresses: addrs}, + BalancerConfig: &BalancerConfig{ReplicationFactor: 7, Spread: 1, Algorithm: AlgorithmRendezvous}, + })) + require.Same(t, before, b.hashring) +} + +// With rendezvous hashing, the picker, the RingView, and a standalone +// rendezvous.Set with the same members all agree, and a failed backend's +// keys move to a remaining one. +func TestRendezvousPickerRoutes(t *testing.T) { + bld := NewBuilder(xxhash.Sum64) + addrs := []resolver.Address{{ServerName: "t", Addr: "1"}, {ServerName: "t", Addr: "2"}} + b := readyBalancerForTarget(t, bld, "test:///backends", addrs...) + require.NoError(t, b.UpdateClientConnState(balancer.ClientConnState{ + ResolverState: resolver.State{Addresses: addrs}, + BalancerConfig: &BalancerConfig{ReplicationFactor: 100, Spread: 1, Algorithm: AlgorithmRendezvous}, + })) + view := bld.RingFor("test:///backends") + + ref := rendezvous.New(xxhash.Sum64) + for _, a := range addrs { + require.NoError(t, ref.Add(subConnMember{key: a.ServerName + a.Addr})) + } + + pick := func(key []byte) balancer.SubConn { + res, err := b.picker.Pick(balancer.PickInfo{Ctx: context.WithValue(context.Background(), CtxKey, key)}) + require.NoError(t, err) + return res.SubConn + } + + var keyFor2 []byte + for i := 0; i < 100; i++ { + key := []byte(fmt.Sprintf("key-%d", i)) + want, err := ref.FindN(key, 1) + require.NoError(t, err) + + viewed, err := view.FindN(key, 1) + require.NoError(t, err) + require.Equal(t, want[0].Key(), viewed[0].Key()) + + picked := pick(key) + require.Equal(t, want[0].Key(), b.scKeys[picked]) + + if want[0].Key() == "t2" { + keyFor2 = key + } + } + require.NotNil(t, keyFor2, "no key hashes to t2") + + sci2, _ := b.subConns.Get(addrs[1]) + b.UpdateSubConnState(sci2.(balancer.SubConn), balancer.SubConnState{ + ConnectivityState: connectivity.TransientFailure, + ConnectionError: fmt.Errorf("refused"), + }) + sci1, _ := b.subConns.Get(addrs[0]) + require.Equal(t, sci1.(balancer.SubConn), pick(keyFor2)) +} diff --git a/balancer.go b/balancer.go index 933ab0a..fecb352 100644 --- a/balancer.go +++ b/balancer.go @@ -26,6 +26,7 @@ import ( "google.golang.org/grpc/serviceconfig" "github.com/authzed/consistent/hashring" + "github.com/authzed/consistent/rendezvous" ) type ctxKey string @@ -49,6 +50,14 @@ const ( // DefaultSpread is the value that will be used when parsing a service // config provides an invalid value. DefaultSpread = 1 + + // AlgorithmHashring places backends on a consistent hashring with + // ReplicationFactor virtual nodes each. This is the default. + AlgorithmHashring = "hashring" + + // AlgorithmRendezvous selects backends with rendezvous (highest random + // weight) hashing. It ignores ReplicationFactor. + AlgorithmRendezvous = "rendezvous" ) // DefaultServiceConfigJSON is a helper to easily leverage the defaults. @@ -73,6 +82,46 @@ type BalancerConfig struct { serviceconfig.LoadBalancingConfig `json:"-"` ReplicationFactor uint16 `json:"replicationFactor,omitempty"` Spread uint8 `json:"spread,omitempty"` + // Algorithm is AlgorithmHashring or AlgorithmRendezvous. Empty means + // AlgorithmHashring. + Algorithm string `json:"algorithm,omitempty"` +} + +func (c *BalancerConfig) algorithm() string { + if c.Algorithm == "" { + return AlgorithmHashring + } + return c.Algorithm +} + +// needsNewMemberSet reports whether moving from the old config to the new one +// requires building a new member set. +func needsNewMemberSet(old, updated *BalancerConfig) bool { + if old == nil || old.algorithm() != updated.algorithm() { + return true + } + return updated.algorithm() == AlgorithmHashring && old.ReplicationFactor != updated.ReplicationFactor +} + +// memberSet is the structure a balancer places its backends on. Both +// hashring.Ring and rendezvous.Set implement it. +type memberSet interface { + Add(hashring.Member) error + Remove(hashring.Member) error + FindN(key []byte, num uint8) ([]hashring.Member, error) + Members() []hashring.Member +} + +var ( + _ memberSet = (*hashring.Ring)(nil) + _ memberSet = (*rendezvous.Set)(nil) +) + +func newMemberSet(hashfn hashring.HashFunc, cfg *BalancerConfig) memberSet { + if cfg.algorithm() == AlgorithmRendezvous { + return rendezvous.New(hashfn) + } + return hashring.MustNew(hashfn, cfg.ReplicationFactor) } // ServiceConfigJSON encodes the current config into the gRPC Service Config @@ -211,6 +260,12 @@ func (b *builder) ParseConfig(js json.RawMessage) (serviceconfig.LoadBalancingCo lbCfg.Spread = DefaultSpread } + switch lbCfg.Algorithm { + case "", AlgorithmHashring, AlgorithmRendezvous: + default: + return nil, fmt.Errorf("consistent-hashring: unknown algorithm %q", lbCfg.Algorithm) + } + return &lbCfg, nil } @@ -228,11 +283,14 @@ type ringBalancer struct { ringMembers map[balancer.SubConn]struct{} config *BalancerConfig - hashring *hashring.Ring + hashring memberSet hasher hashring.HashFunc // slot is where the balancer publishes its live ring for RingView // readers. slot *ringSlot + // published is the entry this balancer stored in slot, so that Close + // withdraws only its own ring. + published *publishedRing resolverErr error // the last error reported by the resolver; cleared on successful resolution connErr error // the last connection error; cleared upon leaving TransientFailure @@ -274,9 +332,10 @@ func (b *ringBalancer) UpdateClientConnState(s balancer.ClientConnState) error { // update the service config if it has changed if s.BalancerConfig != nil { svcConfig := s.BalancerConfig.(*BalancerConfig) - if b.config == nil || svcConfig.ReplicationFactor != b.config.ReplicationFactor { - b.hashring = hashring.MustNew(b.hasher, svcConfig.ReplicationFactor) - b.slot.ring.Store(b.hashring) + if needsNewMemberSet(b.config, svcConfig) { + b.hashring = newMemberSet(b.hasher, svcConfig) + b.published = &publishedRing{b.hashring} + b.slot.ring.Store(b.published) // The new ring starts empty: put every SubConn that has not // failed back on it. b.ringMembers = make(map[balancer.SubConn]struct{}) @@ -488,7 +547,7 @@ func (b *ringBalancer) Close() { // balancer that no longer exists. The CompareAndSwap clears only this // balancer's own ring, so it cannot remove the ring of a second balancer // that shares the target's slot. - b.slot.ring.CompareAndSwap(b.hashring, nil) + b.slot.ring.CompareAndSwap(b.published, nil) } func (b *ringBalancer) ExitIdle() { @@ -498,7 +557,7 @@ func (b *ringBalancer) ExitIdle() { } type picker struct { - hashring *hashring.Ring + hashring memberSet spread uint8 } diff --git a/rendezvous/compare_test.go b/rendezvous/compare_test.go new file mode 100644 index 0000000..434a709 --- /dev/null +++ b/rendezvous/compare_test.go @@ -0,0 +1,191 @@ +package rendezvous + +import ( + "fmt" + "math" + "testing" + + "github.com/cespare/xxhash/v2" + "github.com/stretchr/testify/require" + + "github.com/authzed/consistent/hashring" +) + +// The tests and benchmarks in this file compare rendezvous hashing against the +// vnode hashring. Run the distribution report with: +// +// go test ./rendezvous -run TestCompareDistribution -v +// +// and the benchmarks with: +// +// go test ./rendezvous -run '^$' -bench . -benchmem + +var implementations = []struct { + name string + new func() ringLike +}{ + {"ring-r100", func() ringLike { return hashring.MustNew(xxhash.Sum64, 100) }}, + {"ring-r1000", func() ringLike { return hashring.MustNew(xxhash.Sum64, 1000) }}, + {"hrw", func() ringLike { return New(xxhash.Sum64) }}, +} + +// TestCompareDistribution reports, for each implementation and member count: +// +// - max/mean: the most loaded member's share relative to a perfect 1/n +// - cov: coefficient of variation of per-member load +// - remove moved: fraction of keys that change owner when a member leaves +// (ideal is 1/n) +// - max absorb: the largest share of the departed member's keys taken by a +// single survivor, relative to a perfect 1/(n-1) split +func TestCompareDistribution(t *testing.T) { + if testing.Short() { + t.Skip("distribution report is slow") + } + + const numKeys = 200_000 + ks := keys(numKeys, 42) + + t.Logf("%-11s %5s %9s %7s %13s %11s", "impl", "n", "max/mean", "cov", "remove moved", "max absorb") + for _, n := range []int{3, 5, 10, 30, 100} { + for _, impl := range implementations { + s := impl.new() + ns := nodes(n) + for _, node := range ns { + require.NoError(t, s.Add(node)) + } + + before := owners(t, s, ks) + load := map[string]int{} + for _, o := range before { + load[o]++ + } + maxLoad, sumSq := 0, 0.0 + mean := float64(numKeys) / float64(n) + for _, node := range ns { + l := load[node.Key()] + maxLoad = max(maxLoad, l) + sumSq += (float64(l) - mean) * (float64(l) - mean) + } + cov := math.Sqrt(sumSq/float64(n)) / mean + + // Remove the most loaded member, since that is the worst case for + // how its keys are redistributed. + var departed testNode + for _, node := range ns { + if load[node.Key()] == maxLoad { + departed = node + break + } + } + require.NoError(t, s.Remove(departed)) + after := owners(t, s, ks) + + moved := 0 + absorbed := map[string]int{} + for i := range ks { + if before[i] != after[i] { + moved++ + absorbed[after[i]]++ + } + } + maxAbsorbed := 0 + for _, a := range absorbed { + maxAbsorbed = max(maxAbsorbed, a) + } + fairAbsorb := float64(moved) / float64(n-1) + + t.Logf("%-11s %5d %9.3f %7.3f %13.3f %11.3f", + impl.name, n, float64(maxLoad)/mean, cov, + float64(moved)/numKeys, float64(maxAbsorbed)/fairAbsorb) + } + } +} + +func benchSizes() []int { return []int{3, 10, 30, 100, 1000} } + +// BenchmarkFindN measures a single lookup, which is what the balancer's +// picker does for every dispatched RPC. +func BenchmarkFindN(b *testing.B) { + // Cycle through many keys so the branch predictor cannot memorize the + // path through the lookup. + ks := keys(4096, 7) + + for _, spread := range []uint8{1, 2} { + for _, n := range benchSizes() { + for _, impl := range implementations { + b.Run(fmt.Sprintf("spread=%d/n=%d/%s", spread, n, impl.name), func(b *testing.B) { + s := impl.new() + for _, node := range nodes(n) { + require.NoError(b, s.Add(node)) + } + + b.ReportAllocs() + i := 0 + for b.Loop() { + if _, err := s.FindN(ks[i&(len(ks)-1)], spread); err != nil { + b.Fatal(err) + } + i++ + } + }) + } + } + } +} + +// BenchmarkFindNParallel measures lookups from many goroutines at once, since +// every in-flight dispatch picks concurrently against the same structure. +func BenchmarkFindNParallel(b *testing.B) { + ks := keys(4096, 8) + + for _, n := range []int{10, 30} { + for _, impl := range implementations { + b.Run(fmt.Sprintf("n=%d/%s", n, impl.name), func(b *testing.B) { + s := impl.new() + for _, node := range nodes(n) { + require.NoError(b, s.Add(node)) + } + + b.ReportAllocs() + b.RunParallel(func(pb *testing.PB) { + i := 0 + for pb.Next() { + if _, err := s.FindN(ks[i&(len(ks)-1)], 1); err != nil { + b.Error(err) + return + } + i++ + } + }) + }) + } + } +} + +// BenchmarkMembershipChange measures one member leaving and rejoining, which +// happens whenever a backend's connectivity state changes. +func BenchmarkMembershipChange(b *testing.B) { + // n=1000 is omitted: the ring re-sorts up to a million vnodes per change, + // which takes minutes to benchmark. + for _, n := range []int{3, 10, 30, 100} { + for _, impl := range implementations { + b.Run(fmt.Sprintf("n=%d/%s", n, impl.name), func(b *testing.B) { + s := impl.new() + ns := nodes(n) + for _, node := range ns { + require.NoError(b, s.Add(node)) + } + + b.ReportAllocs() + for b.Loop() { + if err := s.Remove(ns[0]); err != nil { + b.Fatal(err) + } + if err := s.Add(ns[0]); err != nil { + b.Fatal(err) + } + } + }) + } + } +} diff --git a/rendezvous/rendezvous.go b/rendezvous/rendezvous.go new file mode 100644 index 0000000..983f70f --- /dev/null +++ b/rendezvous/rendezvous.go @@ -0,0 +1,193 @@ +// Package rendezvous implements a thread-safe rendezvous (highest random +// weight) hashing member set with a pluggable hashing algorithm. +// +// It exposes the same API shape as the hashring package so that the two can +// be used interchangeably. +package rendezvous + +import ( + "slices" + "strings" + "sync" + + "github.com/authzed/consistent/hashring" +) + +// These are the same values as their hashring counterparts, so callers can +// check errors identically for either implementation. +var ( + ErrMemberAlreadyExists = hashring.ErrMemberAlreadyExists + ErrMemberNotFound = hashring.ErrMemberNotFound + ErrNotEnoughMembers = hashring.ErrNotEnoughMembers +) + +// Set provides a thread-safe rendezvous hashing implementation. +// +// Each lookup hashes the key once and then scores every member by mixing the +// key hash with the member's precomputed hash. The members with the highest +// scores are selected. Lookups are O(n) in the number of members, but each +// score is a handful of arithmetic instructions over a small, contiguous +// slice, which is faster than a vnode ring for the member counts typical of +// a gRPC backend set. +type Set struct { + hashfn hashring.HashFunc + + sync.RWMutex + // members is kept sorted by key so that ties in score are broken + // deterministically, independent of insertion order. + members []member + // hashes holds the hash of members[i] at index i. It is kept separate so + // the scoring loop reads a dense slice. + hashes []uint64 +} + +type member struct { + key string + member hashring.Member +} + +// New allocates a Set with the specified hash function. +func New(hashfn hashring.HashFunc) *Set { + return &Set{hashfn: hashfn} +} + +func cmpKey(m member, key string) int { + return strings.Compare(m.key, key) +} + +// Add inserts a member into the set. +// +// If a member with the same key is already in the set, ErrMemberAlreadyExists +// is returned. +func (s *Set) Add(m hashring.Member) error { + key := m.Key() + hash := s.hashfn([]byte(key)) + + s.Lock() + defer s.Unlock() + + i, found := slices.BinarySearchFunc(s.members, key, cmpKey) + if found { + return ErrMemberAlreadyExists + } + s.members = slices.Insert(s.members, i, member{key, m}) + s.hashes = slices.Insert(s.hashes, i, hash) + return nil +} + +// Remove removes the specified member from the set. +// +// If no member can be found, ErrMemberNotFound is returned. +func (s *Set) Remove(m hashring.Member) error { + key := m.Key() + + s.Lock() + defer s.Unlock() + + i, found := slices.BinarySearchFunc(s.members, key, cmpKey) + if !found { + return ErrMemberNotFound + } + s.members = slices.Delete(s.members, i, i+1) + s.hashes = slices.Delete(s.hashes, i, i+1) + return nil +} + +// FindN returns the N members with the highest scores for the specified key, +// ordered from highest to lowest score. +// +// If there are not enough members to satisfy the request, ErrNotEnoughMembers +// is returned. +func (s *Set) FindN(key []byte, num uint8) ([]hashring.Member, error) { + s.RLock() + defer s.RUnlock() + + if int(num) > len(s.members) { + return nil, ErrNotEnoughMembers + } + if num == 0 { + return []hashring.Member{}, nil + } + + keyHash := s.hashfn(key) + + if num == 1 { + best, bestScore := 0, combineHashes(keyHash, s.hashes[0]) + for i := 1; i < len(s.hashes); i++ { + // Strict comparison: on a tie, the member with the lower key wins. + if sc := combineHashes(keyHash, s.hashes[i]); sc > bestScore { + best, bestScore = i, sc + } + } + return []hashring.Member{s.members[best].member}, nil + } + + // Maintain the top num candidates, sorted descending by score, with an + // insertion sort. num is small, so this beats a heap. + type candidate struct { + score uint64 + index int + } + // A variable-size make stays on the stack only up to 32 bytes (two + // candidates), so use a fixed buffer for common spreads. + var buf [8]candidate + top := buf[:0] + if int(num) > len(buf) { + top = make([]candidate, 0, num) + } + for hashIndex, nodeHash := range s.hashes { + sc := combineHashes(keyHash, nodeHash) + if len(top) == int(num) && sc <= top[len(top)-1].score { + continue + } + if len(top) < int(num) { + top = append(top, candidate{}) + } + // Strict comparison: on a tie, the member with the lower key ranks + // first. + pos := len(top) - 1 + for pos > 0 && top[pos-1].score < sc { + top[pos] = top[pos-1] + pos-- + } + top[pos] = candidate{sc, hashIndex} + } + + found := make([]hashring.Member, len(top)) + for i, c := range top { + found[i] = s.members[c.index].member + } + return found, nil +} + +// Members enumerates the full set of members. +func (s *Set) Members() []hashring.Member { + s.RLock() + defer s.RUnlock() + + membersCopy := make([]hashring.Member, 0, len(s.members)) + for _, m := range s.members { + membersCopy = append(membersCopy, m.member) + } + return membersCopy +} + +// combineHashes combines a key hash and a member hash into the member's weight for +// that key, resulting in a value that can then be sorted for target selection. +// +// The idea is that the resulting value should be relatively randomly and uniformly +// distributed; a naive implementation would use xxhash(old_hash || new_hash) or +// something like that. This function achieves similar results through bitshifting +// and multiplication with some magic constants, referencing the splitmix64 finalizer. +// The method was empirically determined to do a good job of distributing the resulting +// values, and it being bitshifts and multiplications means that it's a good deal faster +// than hashing. Think the fast inverse square root implementation. +func combineHashes(keyHash, memberHash uint64) uint64 { + x := keyHash ^ memberHash + x ^= x >> 30 + x *= 0xbf58476d1ce4e5b9 + x ^= x >> 27 + x *= 0x94d049bb133111eb + x ^= x >> 31 + return x +} diff --git a/rendezvous/rendezvous_test.go b/rendezvous/rendezvous_test.go new file mode 100644 index 0000000..1b07630 --- /dev/null +++ b/rendezvous/rendezvous_test.go @@ -0,0 +1,163 @@ +package rendezvous + +import ( + "encoding/binary" + "math/rand" + "strconv" + "testing" + + "github.com/cespare/xxhash/v2" + "github.com/stretchr/testify/require" + + "github.com/authzed/consistent/hashring" +) + +type testNode string + +func (tn testNode) Key() string { return string(tn) } + +func nodes(n int) []testNode { + out := make([]testNode, n) + for i := range out { + out[i] = testNode("node-" + strconv.Itoa(i)) + } + return out +} + +func keys(n int, seed int64) [][]byte { + r := rand.New(rand.NewSource(seed)) + out := make([][]byte, n) + for i := range out { + out[i] = binary.LittleEndian.AppendUint64(nil, r.Uint64()) + } + return out +} + +func owner(t testing.TB, s *Set, key []byte) string { + found, err := s.FindN(key, 1) + require.NoError(t, err) + return found[0].Key() +} + +func TestAddRemove(t *testing.T) { + s := New(xxhash.Sum64) + + _, err := s.FindN([]byte("key"), 1) + require.ErrorIs(t, err, ErrNotEnoughMembers) + + require.NoError(t, s.Add(testNode("a"))) + require.ErrorIs(t, s.Add(testNode("a")), ErrMemberAlreadyExists) + require.NoError(t, s.Add(testNode("b"))) + require.Len(t, s.Members(), 2) + + found, err := s.FindN([]byte("key"), 0) + require.NoError(t, err) + require.Empty(t, found) + + _, err = s.FindN([]byte("key"), 3) + require.ErrorIs(t, err, ErrNotEnoughMembers) + + require.ErrorIs(t, s.Remove(testNode("c")), ErrMemberNotFound) + require.NoError(t, s.Remove(testNode("a"))) + require.Equal(t, "b", owner(t, s, []byte("key"))) + require.NoError(t, s.Remove(testNode("b"))) + require.Empty(t, s.Members()) +} + +func TestFindNIsOrderedAndDistinct(t *testing.T) { + s := New(xxhash.Sum64) + for _, n := range nodes(10) { + require.NoError(t, s.Add(n)) + } + + for _, key := range keys(1000, 1) { + all, err := s.FindN(key, 10) + require.NoError(t, err) + + seen := map[string]struct{}{} + for _, m := range all { + seen[m.Key()] = struct{}{} + } + require.Len(t, seen, 10) + + // Every prefix of the full ranking must equal FindN with that count. + for num := 1; num <= 10; num++ { + prefix, err := s.FindN(key, uint8(num)) + require.NoError(t, err) + require.Equal(t, all[:num], prefix) + } + } +} + +func TestInsertionOrderIndependence(t *testing.T) { + ns := nodes(20) + a, b := New(xxhash.Sum64), New(xxhash.Sum64) + for _, n := range ns { + require.NoError(t, a.Add(n)) + } + for _, i := range rand.New(rand.NewSource(2)).Perm(len(ns)) { + require.NoError(t, b.Add(ns[i])) + } + + for _, key := range keys(1000, 3) { + fa, err := a.FindN(key, 3) + require.NoError(t, err) + fb, err := b.FindN(key, 3) + require.NoError(t, err) + require.Equal(t, fa, fb) + } +} + +// ringLike is the API shared by hashring.Ring and Set. +type ringLike interface { + Add(hashring.Member) error + Remove(hashring.Member) error + FindN([]byte, uint8) ([]hashring.Member, error) +} + +// TestMinimalDisruption checks that removing a member only moves the keys it +// owned, and adding a member only moves keys onto that member. +func TestMinimalDisruption(t *testing.T) { + impls := map[string]func() ringLike{ + "hrw": func() ringLike { return New(xxhash.Sum64) }, + "ring-r100": func() ringLike { return hashring.MustNew(xxhash.Sum64, 100) }, + } + ks := keys(10_000, 4) + + for name, newImpl := range impls { + t.Run(name, func(t *testing.T) { + s := newImpl() + ns := nodes(10) + for _, n := range ns { + require.NoError(t, s.Add(n)) + } + before := owners(t, s, ks) + + require.NoError(t, s.Remove(ns[3])) + afterRemove := owners(t, s, ks) + for i := range ks { + if before[i] != ns[3].Key() { + require.Equal(t, before[i], afterRemove[i]) + } + } + + require.NoError(t, s.Add(testNode("new"))) + afterAdd := owners(t, s, ks) + for i := range ks { + if afterAdd[i] != "new" { + require.Equal(t, afterRemove[i], afterAdd[i]) + } + } + }) + } +} + +func owners(t testing.TB, s ringLike, ks [][]byte) []string { + out := make([]string, len(ks)) + for i, k := range ks { + found, err := s.FindN(k, 1) + require.NoError(t, err) + out[i] = found[0].Key() + } + return out +} diff --git a/ringview.go b/ringview.go index a32c030..aa0d823 100644 --- a/ringview.go +++ b/ringview.go @@ -37,7 +37,13 @@ type RingView interface { // balancer swaps the pointer when it replaces the ring and clears the // pointer on Close. type ringSlot struct { - ring atomic.Pointer[hashring.Ring] + ring atomic.Pointer[publishedRing] +} + +// publishedRing boxes a memberSet so that ringSlot can hold either +// implementation behind an atomic pointer. +type publishedRing struct { + memberSet } var _ RingView = (*ringSlot)(nil)