Add mutex protection to ConsistentHashRing for thread safety
All checks were successful
CI / build (pull_request) Successful in 17s
All checks were successful
CI / build (pull_request) Successful in 17s
- Add sync.RWMutex to ConsistentHashRing struct - Use Lock/Unlock for write operations (AddNode, RemoveNode) - Use RLock/RUnlock for read operations (GetNode, GetNodes, IsEmpty) This allows concurrent reads (the common case) while serializing writes, preventing race conditions when multiple goroutines access the hash ring. Closes #35 Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
@@ -5,10 +5,12 @@ import (
|
|||||||
"encoding/binary"
|
"encoding/binary"
|
||||||
"fmt"
|
"fmt"
|
||||||
"sort"
|
"sort"
|
||||||
|
"sync"
|
||||||
)
|
)
|
||||||
|
|
||||||
// ConsistentHashRing implements a consistent hash ring for shard distribution
|
// ConsistentHashRing implements a consistent hash ring for shard distribution
|
||||||
type ConsistentHashRing struct {
|
type ConsistentHashRing struct {
|
||||||
|
mu sync.RWMutex
|
||||||
ring map[uint32]string // hash -> node ID
|
ring map[uint32]string // hash -> node ID
|
||||||
sortedHashes []uint32 // sorted hash keys
|
sortedHashes []uint32 // sorted hash keys
|
||||||
nodes map[string]bool // active nodes
|
nodes map[string]bool // active nodes
|
||||||
@@ -24,6 +26,9 @@ func NewConsistentHashRing() *ConsistentHashRing {
|
|||||||
|
|
||||||
// AddNode adds a node to the hash ring
|
// AddNode adds a node to the hash ring
|
||||||
func (chr *ConsistentHashRing) AddNode(nodeID string) {
|
func (chr *ConsistentHashRing) AddNode(nodeID string) {
|
||||||
|
chr.mu.Lock()
|
||||||
|
defer chr.mu.Unlock()
|
||||||
|
|
||||||
if chr.nodes[nodeID] {
|
if chr.nodes[nodeID] {
|
||||||
return // Node already exists
|
return // Node already exists
|
||||||
}
|
}
|
||||||
@@ -45,6 +50,9 @@ func (chr *ConsistentHashRing) AddNode(nodeID string) {
|
|||||||
|
|
||||||
// RemoveNode removes a node from the hash ring
|
// RemoveNode removes a node from the hash ring
|
||||||
func (chr *ConsistentHashRing) RemoveNode(nodeID string) {
|
func (chr *ConsistentHashRing) RemoveNode(nodeID string) {
|
||||||
|
chr.mu.Lock()
|
||||||
|
defer chr.mu.Unlock()
|
||||||
|
|
||||||
if !chr.nodes[nodeID] {
|
if !chr.nodes[nodeID] {
|
||||||
return // Node doesn't exist
|
return // Node doesn't exist
|
||||||
}
|
}
|
||||||
@@ -65,6 +73,9 @@ func (chr *ConsistentHashRing) RemoveNode(nodeID string) {
|
|||||||
|
|
||||||
// GetNode returns the node responsible for a given key
|
// GetNode returns the node responsible for a given key
|
||||||
func (chr *ConsistentHashRing) GetNode(key string) string {
|
func (chr *ConsistentHashRing) GetNode(key string) string {
|
||||||
|
chr.mu.RLock()
|
||||||
|
defer chr.mu.RUnlock()
|
||||||
|
|
||||||
if len(chr.sortedHashes) == 0 {
|
if len(chr.sortedHashes) == 0 {
|
||||||
return ""
|
return ""
|
||||||
}
|
}
|
||||||
@@ -92,6 +103,9 @@ func (chr *ConsistentHashRing) hash(key string) uint32 {
|
|||||||
|
|
||||||
// GetNodes returns all active nodes in the ring
|
// GetNodes returns all active nodes in the ring
|
||||||
func (chr *ConsistentHashRing) GetNodes() []string {
|
func (chr *ConsistentHashRing) GetNodes() []string {
|
||||||
|
chr.mu.RLock()
|
||||||
|
defer chr.mu.RUnlock()
|
||||||
|
|
||||||
nodes := make([]string, 0, len(chr.nodes))
|
nodes := make([]string, 0, len(chr.nodes))
|
||||||
for nodeID := range chr.nodes {
|
for nodeID := range chr.nodes {
|
||||||
nodes = append(nodes, nodeID)
|
nodes = append(nodes, nodeID)
|
||||||
@@ -101,5 +115,8 @@ func (chr *ConsistentHashRing) GetNodes() []string {
|
|||||||
|
|
||||||
// IsEmpty returns true if the ring has no nodes
|
// IsEmpty returns true if the ring has no nodes
|
||||||
func (chr *ConsistentHashRing) IsEmpty() bool {
|
func (chr *ConsistentHashRing) IsEmpty() bool {
|
||||||
|
chr.mu.RLock()
|
||||||
|
defer chr.mu.RUnlock()
|
||||||
|
|
||||||
return len(chr.nodes) == 0
|
return len(chr.nodes) == 0
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user