Repository: puzpuzpuz/xsync
Pull request #63 (2022-10-27 19:12:38)

Fix duplicate keys on intensive map insertion and deletion


Patch:
diff --git a/map.go b/map.go
index 0753ee7..a84a791 100644
--- a/map.go
+++ b/map.go
@@ -236,8 +236,8 @@ func (m *Map) doStore(key string, valueFn func() interface{}, loadIfExists bool)
 	for {
 	store_attempt:
 		var (
-			emptykp, emptyvp *unsafe.Pointer
-			emptyidx         int
+			emptyb   *bucketPadded
+			emptyidx int
 		)
 		table := (*mapTable)(atomic.LoadPointer(&m.table))
 		tableLen := len(table.buckets)
@@ -261,9 +261,8 @@ func (m *Map) doStore(key string, valueFn func() interface{}, loadIfExists bool)
 			topHashes := atomic.LoadUint64(&b.topHashMutex)
 			for i := 0; i < entriesPerMapBucket; i++ {
 				if b.keys[i] == nil {
-					if emptykp == nil {
-						emptykp = &b.keys[i]
-						emptyvp = &b.values[i]
+					if emptyb == nil {
+						emptyb = b
 						emptyidx = i
 					}
 					continue
@@ -293,13 +292,14 @@ func (m *Map) doStore(key string, valueFn func() interface{}, loadIfExists bool)
 				}
 			}
 			if b.next == nil {
-				if emptykp != nil {
+				if emptyb != nil {
 					// Insertion case. First we update the value, then the key.
 					// This is important for atomic snapshot states.
-					atomic.StoreUint64(&b.topHashMutex, storeTopHash(hash, topHashes, emptyidx))
+					topHashes = atomic.LoadUint64(&emptyb.topHashMutex)
+					atomic.StoreUint64(&emptyb.topHashMutex, storeTopHash(hash, topHashes, emptyidx))
 					value := valueFn()
-					atomic.StorePointer(emptyvp, unsafe.Pointer(&value))
-					atomic.StorePointer(emptykp, unsafe.Pointer(&key))
+					atomic.StorePointer(&emptyb.values[emptyidx], unsafe.Pointer(&value))
+					atomic.StorePointer(&emptyb.keys[emptyidx], unsafe.Pointer(&key))
 					unlockBucket(&rootb.topHashMutex)
 					table.addSize(bidx, 1)
 					return value, false
@@ -316,7 +316,7 @@ func (m *Map) doStore(key string, valueFn func() interface{}, loadIfExists bool)
 				newb.keys[0] = unsafe.Pointer(&key)
 				value := valueFn()
 				newb.values[0] = unsafe.Pointer(&value)
-				newb.topHashMutex = storeTopHash(hash, topHashes, emptyidx)
+				newb.topHashMutex = storeTopHash(hash, newb.topHashMutex, 0)
 				atomic.StorePointer(&b.next, unsafe.Pointer(newb))
 				unlockBucket(&rootb.topHashMutex)
 				table.addSize(bidx, 1)
@@ -430,7 +430,7 @@ func appendToBucket(hash uint64, keyPtr, valPtr unsafe.Pointer, b *bucketPadded)
 			newb := new(bucketPadded)
 			newb.keys[0] = keyPtr
 			newb.values[0] = valPtr
-			newb.topHashMutex = storeTopHash(hash, b.topHashMutex, 0)
+			newb.topHashMutex = storeTopHash(hash, newb.topHashMutex, 0)
 			b.next = unsafe.Pointer(newb)
 			return
 		}
diff --git a/mapof.go b/mapof.go
index f10098f..adf4134 100644
--- a/mapof.go
+++ b/mapof.go
@@ -166,8 +166,8 @@ func (m *MapOf[K, V]) doStore(key K, valueFn func() V, loadIfExists bool) (V, bo
 	for {
 	store_attempt:
 		var (
-			emptykp, emptyvp *unsafe.Pointer
-			emptyidx         int
+			emptyb   *bucketPadded
+			emptyidx int
 		)
 		table := (*mapTable)(atomic.LoadPointer(&m.table))
 		tableLen := len(table.buckets)
@@ -191,9 +191,8 @@ func (m *MapOf[K, V]) doStore(key K, valueFn func() V, loadIfExists bool) (V, bo
 			topHashes := atomic.LoadUint64(&b.topHashMutex)
 			for i := 0; i < entriesPerMapBucket; i++ {
 				if b.keys[i] == nil {
-					if emptykp == nil {
-						emptykp = &b.keys[i]
-						emptyvp = &b.values[i]
+					if emptyb == nil {
+						emptyb = b
 						emptyidx = i
 					}
 					continue
@@ -224,14 +223,15 @@ func (m *MapOf[K, V]) doStore(key K, valueFn func() V, loadIfExists bool) (V, bo
 				}
 			}
 			if b.next == nil {
-				if emptykp != nil {
+				if emptyb != nil {
 					// Insertion case. First we update the value, then the key.
 					// This is important for atomic snapshot states.
-					atomic.StoreUint64(&b.topHashMutex, storeTopHash(hash, topHashes, emptyidx))
+					topHashes = atomic.LoadUint64(&emptyb.topHashMutex)
+					atomic.StoreUint64(&emptyb.topHashMutex, storeTopHash(hash, topHashes, emptyidx))
 					value := valueFn()
 					var wv interface{} = valueFn()
-					atomic.StorePointer(emptyvp, unsafe.Pointer(&wv))
-					atomic.StorePointer(emptykp, unsafe.Pointer(&key))
+					atomic.StorePointer(&emptyb.values[emptyidx], unsafe.Pointer(&wv))
+					atomic.StorePointer(&emptyb.keys[emptyidx], unsafe.Pointer(&key))
 					unlockBucket(&rootb.topHashMutex)
 					table.addSize(bidx, 1)
 					return value, false
@@ -249,7 +249,7 @@ func (m *MapOf[K, V]) doStore(key K, valueFn func() V, loadIfExists bool) (V, bo
 				value := valueFn()
 				var wv interface{} = value
 				newb.values[0] = unsafe.Pointer(&wv)
-				newb.topHashMutex = storeTopHash(hash, topHashes, emptyidx)
+				newb.topHashMutex = storeTopHash(hash, newb.topHashMutex, 0)
 				atomic.StorePointer(&b.next, unsafe.Pointer(newb))
 				unlockBucket(&rootb.topHashMutex)
 				table.addSize(bidx, 1)


Test patch:
diff --git a/map_test.go b/map_test.go
index 3ba82e5..fff8b0e 100644
--- a/map_test.go
+++ b/map_test.go
@@ -468,7 +468,7 @@ func parallelSeqResizer(t *testing.T, m *Map, numEntries int, positive bool, cdo
 	cdone <- true
 }
 
-func TestMapParallelResizeGrowOnly(t *testing.T) {
+func TestMapParallelResize_GrowOnly(t *testing.T) {
 	const numEntries = 100_000
 	m := NewMap()
 	cdone := make(chan bool)
@@ -494,13 +494,13 @@ func TestMapParallelResizeGrowOnly(t *testing.T) {
 
 func parallelRandResizer(t *testing.T, m *Map, numIters, numEntries int, cdone chan bool) {
 	r := rand.New(rand.NewSource(time.Now().UnixNano()))
-	for i := 0; i < numEntries; i++ {
+	for i := 0; i < numIters; i++ {
 		coin := r.Int63n(2)
 		for j := 0; j < numEntries; j++ {
 			if coin == 1 {
-				m.Store(strconv.Itoa(i), i)
+				m.Store(strconv.Itoa(j), j)
 			} else {
-				m.Delete(strconv.Itoa(i))
+				m.Delete(strconv.Itoa(j))
 			}
 		}
 	}
@@ -508,8 +508,8 @@ func parallelRandResizer(t *testing.T, m *Map, numIters, numEntries int, cdone c
 }
 
 func TestMapParallelResize(t *testing.T) {
-	const numIters = 100
-	const numEntries = 1_000
+	const numIters = 1000
+	const numEntries = 2 * EntriesPerMapBucket * MinMapTableLen
 	m := NewMap()
 	cdone := make(chan bool)
 	go parallelRandResizer(t, m, numIters, numEntries, cdone)
@@ -528,9 +528,18 @@ func TestMapParallelResize(t *testing.T) {
 			t.Errorf("values do not match for %d: %v", i, v)
 		}
 	}
-	if s := m.Size(); s > numEntries {
+	s := m.Size()
+	if s > numEntries {
 		t.Errorf("unexpected size: %v", s)
 	}
+	rs := 0
+	m.Range(func(key string, value interface{}) bool {
+		rs++
+		return true
+	})
+	if s != rs {
+		t.Errorf("size does not match number of entries in Range: %v, %v", s, rs)
+	}
 }
 
 func parallelSeqStorer(t *testing.T, m *Map, storeEach, numIters, numEntries int, cdone chan bool) {
diff --git a/mapof_test.go b/mapof_test.go
index 88d2644..ac8b10f 100644
--- a/mapof_test.go
+++ b/mapof_test.go
@@ -606,7 +606,7 @@ func parallelSeqTypedResizer(t *testing.T, m *MapOf[int, int], numEntries int, p
 	cdone <- true
 }
 
-func TestMapOfParallelResizeGrowOnly(t *testing.T) {
+func TestMapOfParallelResize_GrowOnly(t *testing.T) {
 	const numEntries = 100_000
 	m := NewIntegerMapOf[int, int]()
 	cdone := make(chan bool)
@@ -632,13 +632,13 @@ func TestMapOfParallelResizeGrowOnly(t *testing.T) {
 
 func parallelRandTypedResizer(t *testing.T, m *MapOf[int, int], numIters, numEntries int, cdone chan bool) {
 	r := rand.New(rand.NewSource(time.Now().UnixNano()))
-	for i := 0; i < numEntries; i++ {
+	for i := 0; i < numIters; i++ {
 		coin := r.Int63n(2)
 		for j := 0; j < numEntries; j++ {
 			if coin == 1 {
-				m.Store(i, i)
+				m.Store(j, j)
 			} else {
-				m.Delete(i)
+				m.Delete(j)
 			}
 		}
 	}
@@ -646,8 +646,8 @@ func parallelRandTypedResizer(t *testing.T, m *MapOf[int, int], numIters, numEnt
 }
 
 func TestMapOfParallelResize(t *testing.T) {
-	const numIters = 100
-	const numEntries = 1_000
+	const numIters = 1000
+	const numEntries = 2 * EntriesPerMapBucket * MinMapTableLen
 	m := NewIntegerMapOf[int, int]()
 	cdone := make(chan bool)
 	go parallelRandTypedResizer(t, m, numIters, numEntries, cdone)
@@ -666,9 +666,18 @@ func TestMapOfParallelResize(t *testing.T) {
 			t.Errorf("values do not match for %d: %v", i, v)
 		}
 	}
-	if s := m.Size(); s > numEntries {
+	s := m.Size()
+	if s > numEntries {
 		t.Errorf("unexpected size: %v", s)
 	}
+	rs := 0
+	m.Range(func(key int, value int) bool {
+		rs++
+		return true
+	})
+	if s != rs {
+		t.Errorf("size does not match number of entries in Range: %v, %v", s, rs)
+	}
 }
 
 func parallelSeqTypedStorer(t *testing.T, m *MapOf[string, int], storeEach, numIters, numEntries int, cdone chan bool) {