core/txpool: protect cache with mutex #27898, close XFN-65 (#1630)

core/txpool/list.go

@@ -52,9 +52,10 @@ func (h *nonceHeap) Pop() interface{} {
 // sortedMap is a nonce->transaction hash map with a heap based index to allow
 // iterating over the contents in a nonce-incrementing way.
 type sortedMap struct {
-	items map[uint64]*types.Transaction // Hash map storing the transaction data
-	index *nonceHeap                    // Heap of nonces of all the stored transactions (non-strict mode)
-	cache types.Transactions            // Cache of the transactions already sorted
+	items   map[uint64]*types.Transaction // Hash map storing the transaction data
+	index   *nonceHeap                    // Heap of nonces of all the stored transactions (non-strict mode)
+	cache   types.Transactions            // Cache of the transactions already sorted
+	cacheMu sync.Mutex                    // Mutex covering the cache
 }
 
 // newSortedMap creates a new nonce-sorted transaction map.
@@ -77,7 +78,9 @@ func (m *sortedMap) Put(tx *types.Transaction) {
 	if m.items[nonce] == nil {
 		heap.Push(m.index, nonce)
 	}
+	m.cacheMu.Lock()
 	m.items[nonce], m.cache = tx, nil
+	m.cacheMu.Unlock()
 }
 
 // Forward removes all transactions from the map with a nonce lower than the
@@ -93,9 +96,11 @@ func (m *sortedMap) Forward(threshold uint64) types.Transactions {
 		delete(m.items, nonce)
 	}
 	// If we had a cached order, shift the front
+	m.cacheMu.Lock()
 	if m.cache != nil {
 		m.cache = m.cache[len(removed):]
 	}
+	m.cacheMu.Unlock()
 	return removed
 }
 
@@ -119,7 +124,9 @@ func (m *sortedMap) reheap() {
 		*m.index = append(*m.index, nonce)
 	}
 	heap.Init(m.index)
+	m.cacheMu.Lock()
 	m.cache = nil
+	m.cacheMu.Unlock()
 }
 
 // filter is identical to Filter, but **does not** regenerate the heap. This method
@@ -135,7 +142,9 @@ func (m *sortedMap) filter(filterFunc func(*types.Transaction) bool) types.Trans
 		}
 	}
 	if len(removed) > 0 {
+		m.cacheMu.Lock()
 		m.cache = nil
+		m.cacheMu.Unlock()
 	}
 	return removed
 }
@@ -159,9 +168,11 @@ func (m *sortedMap) Cap(threshold int) types.Transactions {
 	heap.Init(m.index)
 
 	// If we had a cache, shift the back
+	m.cacheMu.Lock()
 	if m.cache != nil {
 		m.cache = m.cache[:len(m.cache)-len(drops)]
 	}
+	m.cacheMu.Unlock()
 	return drops
 }
 
@@ -181,7 +192,9 @@ func (m *sortedMap) Remove(nonce uint64) bool {
 		}
 	}
 	delete(m.items, nonce)
+	m.cacheMu.Lock()
 	m.cache = nil
+	m.cacheMu.Unlock()
 
 	return true
 }
@@ -205,7 +218,9 @@ func (m *sortedMap) Ready(start uint64) types.Transactions {
 		delete(m.items, next)
 		heap.Pop(m.index)
 	}
+	m.cacheMu.Lock()
 	m.cache = nil
+	m.cacheMu.Unlock()
 
 	return ready
 }
@@ -216,6 +231,8 @@ func (m *sortedMap) Len() int {
 }
 
 func (m *sortedMap) flatten() types.Transactions {
+	m.cacheMu.Lock()
+	defer m.cacheMu.Unlock()
 	// If the sorting was not cached yet, create and cache it
 	if m.cache == nil {
 		m.cache = make(types.Transactions, 0, len(m.items))
@@ -231,8 +248,8 @@ func (m *sortedMap) flatten() types.Transactions {
 // sorted internal representation. The result of the sorting is cached in case
 // it's requested again before any modifications are made to the contents.
 func (m *sortedMap) Flatten() types.Transactions {
-	// Copy the cache to prevent accidental modifications
 	cache := m.flatten()
+	// Copy the cache to prevent accidental modification
 	txs := make(types.Transactions, len(cache))
 	copy(txs, cache)
 	return txs