345 lines
8.3 KiB
Go
345 lines
8.3 KiB
Go
package main
|
||
|
||
import (
|
||
"encoding/binary"
|
||
"fmt"
|
||
"hash/fnv"
|
||
"sort"
|
||
"strings"
|
||
"sync"
|
||
)
|
||
|
||
// ring.go
|
||
//
|
||
// Consistent hashing ring skeleton for future integration.
|
||
// --------------------------------------------------------
|
||
// This file introduces a minimal, allocation‑light consistent hashing structure
|
||
// intended to replace per-cart ownership negotiation. It focuses on:
|
||
// * Deterministic lookup: O(log V) via binary search
|
||
// * Even(ish) distribution using virtual nodes (vnodes)
|
||
// * Epoch / fingerprint tracking to detect membership drift
|
||
//
|
||
// NOT YET WIRED:
|
||
// * SyncedPool integration (ownerForCart, lazy migration)
|
||
// * Replication factor > 1
|
||
// * Persistent state migration
|
||
//
|
||
// Safe to import now; unused until explicit integration code is added.
|
||
//
|
||
// Design Notes
|
||
// ------------
|
||
// - Hosts contribute `vnodesPerHost` virtual nodes. Higher counts smooth
|
||
// distribution at cost of memory (V = hosts * vnodesPerHost).
|
||
// - Hash of vnode = FNV1a64(host + "#" + index). For improved quality you
|
||
// can swap in xxhash or siphash later without changing API (but doing so
|
||
// will reshuffle ownership).
|
||
// - Cart ownership lookup uses either cartID.Raw() when provided (uniform
|
||
// 64-bit space) or falls back to hashing string forms (legacy).
|
||
// - Epoch is monotonically increasing; consumers can fence stale results.
|
||
//
|
||
// Future Extensions
|
||
// -----------------
|
||
// - Weighted hosts (proportionally more vnodes).
|
||
// - Replication: LookupN(h, n) to return primary + replicas.
|
||
// - Streaming / diff-based ring updates (gossip).
|
||
// - Hash function injection for deterministic test scenarios.
|
||
//
|
||
// ---------------------------------------------------------------------------
|
||
|
||
// Vnode represents a single virtual node position on the ring.
|
||
type Vnode struct {
|
||
Hash uint64 // position on the ring
|
||
Host string // physical host owning this vnode
|
||
Index int // per-host vnode index (0..vnodesPerHost-1)
|
||
}
|
||
|
||
// Ring is an immutable consistent hash ring snapshot.
|
||
type Ring struct {
|
||
Epoch uint64
|
||
Vnodes []Vnode // sorted by Hash
|
||
hosts []string
|
||
fingerprint uint64 // membership fingerprint (order-independent)
|
||
}
|
||
|
||
// RingBuilder accumulates parameters to construct a Ring.
|
||
type RingBuilder struct {
|
||
epoch uint64
|
||
vnodesPerHost int
|
||
hosts []string
|
||
}
|
||
|
||
// NewRingBuilder creates a builder with defaults.
|
||
func NewRingBuilder() *RingBuilder {
|
||
return &RingBuilder{
|
||
vnodesPerHost: 64, // a reasonable default for small clusters
|
||
}
|
||
}
|
||
|
||
func (b *RingBuilder) WithEpoch(e uint64) *RingBuilder {
|
||
b.epoch = e
|
||
return b
|
||
}
|
||
|
||
func (b *RingBuilder) WithVnodesPerHost(n int) *RingBuilder {
|
||
if n > 0 {
|
||
b.vnodesPerHost = n
|
||
}
|
||
return b
|
||
}
|
||
|
||
func (b *RingBuilder) WithHosts(hosts []string) *RingBuilder {
|
||
uniq := make(map[string]struct{}, len(hosts))
|
||
out := make([]string, 0, len(hosts))
|
||
for _, h := range hosts {
|
||
h = strings.TrimSpace(h)
|
||
if h == "" {
|
||
continue
|
||
}
|
||
if _, ok := uniq[h]; ok {
|
||
continue
|
||
}
|
||
uniq[h] = struct{}{}
|
||
out = append(out, h)
|
||
}
|
||
sort.Strings(out)
|
||
b.hosts = out
|
||
return b
|
||
}
|
||
|
||
func (b *RingBuilder) Build() *Ring {
|
||
if len(b.hosts) == 0 {
|
||
return &Ring{
|
||
Epoch: b.epoch,
|
||
Vnodes: nil,
|
||
hosts: nil,
|
||
fingerprint: 0,
|
||
}
|
||
}
|
||
|
||
totalVnodes := len(b.hosts) * b.vnodesPerHost
|
||
vnodes := make([]Vnode, 0, totalVnodes)
|
||
|
||
for _, host := range b.hosts {
|
||
for i := 0; i < b.vnodesPerHost; i++ {
|
||
h := hashVnode(host, i)
|
||
vnodes = append(vnodes, Vnode{
|
||
Hash: h,
|
||
Host: host,
|
||
Index: i,
|
||
})
|
||
}
|
||
}
|
||
sort.Slice(vnodes, func(i, j int) bool {
|
||
if vnodes[i].Hash == vnodes[j].Hash {
|
||
// Tie-break deterministically by host then index to avoid instability
|
||
if vnodes[i].Host == vnodes[j].Host {
|
||
return vnodes[i].Index < vnodes[j].Index
|
||
}
|
||
return vnodes[i].Host < vnodes[j].Host
|
||
}
|
||
return vnodes[i].Hash < vnodes[j].Hash
|
||
})
|
||
|
||
fp := fingerprintHosts(b.hosts)
|
||
|
||
return &Ring{
|
||
Epoch: b.epoch,
|
||
Vnodes: vnodes,
|
||
hosts: append([]string(nil), b.hosts...),
|
||
fingerprint: fp,
|
||
}
|
||
}
|
||
|
||
// Hosts returns a copy of the host list (sorted).
|
||
func (r *Ring) Hosts() []string {
|
||
if len(r.hosts) == 0 {
|
||
return nil
|
||
}
|
||
cp := make([]string, len(r.hosts))
|
||
copy(cp, r.hosts)
|
||
return cp
|
||
}
|
||
|
||
// Fingerprint returns a hash representing the unordered membership set.
|
||
func (r *Ring) Fingerprint() uint64 {
|
||
return r.fingerprint
|
||
}
|
||
|
||
// Empty indicates ring has no vnodes.
|
||
func (r *Ring) Empty() bool {
|
||
return len(r.Vnodes) == 0
|
||
}
|
||
|
||
// Lookup returns the vnode owning a given hash value.
|
||
func (r *Ring) Lookup(h uint64) Vnode {
|
||
if len(r.Vnodes) == 0 {
|
||
return Vnode{}
|
||
}
|
||
// Binary search: first position with Hash >= h
|
||
i := sort.Search(len(r.Vnodes), func(i int) bool {
|
||
return r.Vnodes[i].Hash >= h
|
||
})
|
||
if i == len(r.Vnodes) {
|
||
return r.Vnodes[0]
|
||
}
|
||
return r.Vnodes[i]
|
||
}
|
||
|
||
// LookupID selects owner vnode for a CartID (fast path).
|
||
func (r *Ring) LookupID(id CartID) Vnode {
|
||
return r.Lookup(id.Raw())
|
||
}
|
||
|
||
// LookupString hashes an arbitrary string and looks up owner.
|
||
func (r *Ring) LookupString(s string) Vnode {
|
||
return r.Lookup(hashKeyString(s))
|
||
}
|
||
|
||
// LookupN returns up to n distinct host vnodes in ring order
|
||
// starting from the primary owner of hash h (for replication).
|
||
func (r *Ring) LookupN(h uint64, n int) []Vnode {
|
||
if n <= 0 || len(r.Vnodes) == 0 {
|
||
return nil
|
||
}
|
||
if n > len(r.hosts) {
|
||
n = len(r.hosts)
|
||
}
|
||
owners := make([]Vnode, 0, n)
|
||
seen := make(map[string]struct{}, n)
|
||
|
||
start := r.Lookup(h)
|
||
|
||
// Find index of start (can binary search again or linear scan; since we
|
||
// already have start.Hash we do another search for clarity)
|
||
i := sort.Search(len(r.Vnodes), func(i int) bool {
|
||
return r.Vnodes[i].Hash >= start.Hash
|
||
})
|
||
if i == len(r.Vnodes) {
|
||
i = 0
|
||
}
|
||
|
||
for idx := 0; len(owners) < n && idx < len(r.Vnodes); idx++ {
|
||
v := r.Vnodes[(i+idx)%len(r.Vnodes)]
|
||
if _, ok := seen[v.Host]; ok {
|
||
continue
|
||
}
|
||
seen[v.Host] = struct{}{}
|
||
owners = append(owners, v)
|
||
}
|
||
return owners
|
||
}
|
||
|
||
// DiffHosts compares this ring's membership to another.
|
||
func (r *Ring) DiffHosts(other *Ring) (added []string, removed []string) {
|
||
if other == nil {
|
||
return r.Hosts(), nil
|
||
}
|
||
cur := make(map[string]struct{}, len(r.hosts))
|
||
for _, h := range r.hosts {
|
||
cur[h] = struct{}{}
|
||
}
|
||
oth := make(map[string]struct{}, len(other.hosts))
|
||
for _, h := range other.hosts {
|
||
oth[h] = struct{}{}
|
||
}
|
||
for h := range cur {
|
||
if _, ok := oth[h]; !ok {
|
||
removed = append(removed, h)
|
||
}
|
||
}
|
||
for h := range oth {
|
||
if _, ok := cur[h]; !ok {
|
||
added = append(added, h)
|
||
}
|
||
}
|
||
sort.Strings(added)
|
||
sort.Strings(removed)
|
||
return
|
||
}
|
||
|
||
// ---------------------------- Hash Functions ---------------------------------
|
||
|
||
func hashVnode(host string, idx int) uint64 {
|
||
h := fnv.New64a()
|
||
_, _ = h.Write([]byte(host))
|
||
_, _ = h.Write([]byte{'#'})
|
||
var buf [8]byte
|
||
binary.BigEndian.PutUint64(buf[:], uint64(idx))
|
||
_, _ = h.Write(buf[:])
|
||
return h.Sum64()
|
||
}
|
||
|
||
// hashKeyString provides a stable hash for arbitrary string keys (legacy IDs).
|
||
func hashKeyString(s string) uint64 {
|
||
h := fnv.New64a()
|
||
_, _ = h.Write([]byte(s))
|
||
return h.Sum64()
|
||
}
|
||
|
||
// fingerprintHosts produces an order-insensitive hash over the host set.
|
||
func fingerprintHosts(hosts []string) uint64 {
|
||
if len(hosts) == 0 {
|
||
return 0
|
||
}
|
||
h := fnv.New64a()
|
||
for _, host := range hosts {
|
||
_, _ = h.Write([]byte(host))
|
||
_, _ = h.Write([]byte{0})
|
||
}
|
||
return h.Sum64()
|
||
}
|
||
|
||
// --------------------------- Thread-Safe Wrapper -----------------------------
|
||
//
|
||
// RingRef offers atomic swap + read semantics. SyncedPool can embed or hold
|
||
// one of these to manage live ring updates safely.
|
||
|
||
type RingRef struct {
|
||
mu sync.RWMutex
|
||
ring *Ring
|
||
}
|
||
|
||
func NewRingRef(r *Ring) *RingRef {
|
||
return &RingRef{ring: r}
|
||
}
|
||
|
||
func (rr *RingRef) Get() *Ring {
|
||
rr.mu.RLock()
|
||
r := rr.ring
|
||
rr.mu.RUnlock()
|
||
return r
|
||
}
|
||
|
||
func (rr *RingRef) Set(r *Ring) {
|
||
rr.mu.Lock()
|
||
rr.ring = r
|
||
rr.mu.Unlock()
|
||
}
|
||
|
||
func (rr *RingRef) LookupID(id CartID) Vnode {
|
||
r := rr.Get()
|
||
if r == nil {
|
||
return Vnode{}
|
||
}
|
||
return r.LookupID(id)
|
||
}
|
||
|
||
// ----------------------------- Debug Utilities -------------------------------
|
||
|
||
func (r *Ring) String() string {
|
||
var b strings.Builder
|
||
fmt.Fprintf(&b, "Ring{epoch=%d vnodes=%d hosts=%d}\n", r.Epoch, len(r.Vnodes), len(r.hosts))
|
||
limit := len(r.Vnodes)
|
||
if limit > 16 {
|
||
limit = 16
|
||
}
|
||
for i := 0; i < limit; i++ {
|
||
v := r.Vnodes[i]
|
||
fmt.Fprintf(&b, " %02d hash=%016x host=%s idx=%d\n", i, v.Hash, v.Host, v.Index)
|
||
}
|
||
if len(r.Vnodes) > limit {
|
||
fmt.Fprintf(&b, " ... (%d more)\n", len(r.Vnodes)-limit)
|
||
}
|
||
return b.String()
|
||
}
|