clean-up mutexes

Author: julienrbrt

block/internal/cache/generic_cache.go

@@ -31,22 +31,24 @@ const (
 
 // Cache is a generic cache that maintains items that are seen and hard confirmed.
 // Uses bounded LRU caches to prevent unbounded memory growth.
+// The underlying LRU caches are thread-safe. Mutexes are only used where
+// compound operations require atomicity.
 type Cache[T any] struct {
-	// itemsByHeight stores items keyed by uint64 height
+	// itemsByHeight stores items keyed by uint64 height.
+	// Mutex needed for atomic get-and-remove in getNextItem.
 	itemsByHeight   *lru.Cache[uint64, *T]
-	itemsByHeightMu sync.RWMutex
+	itemsByHeightMu sync.Mutex
 
 	// hashes tracks whether a given hash has been seen
-	hashes   *lru.Cache[string, bool]
-	hashesMu sync.RWMutex
+	hashes *lru.Cache[string, bool]
 
 	// daIncluded tracks the DA inclusion height for a given hash
-	daIncluded   *lru.Cache[string, uint64]
-	daIncludedMu sync.RWMutex
+	daIncluded *lru.Cache[string, uint64]
 
-	// hashByHeight tracks the hash associated with each height for pruning
+	// hashByHeight tracks the hash associated with each height for pruning.
+	// Mutex needed for atomic operations in deleteAllForHeight.
 	hashByHeight   *lru.Cache[uint64, string]
-	hashByHeightMu sync.RWMutex
+	hashByHeightMu sync.Mutex
 
 	// maxDAHeight tracks the maximum DA height seen
 	maxDAHeight *atomic.Uint64
@@ -111,9 +113,6 @@ func NewCacheWithConfig[T any](config CacheConfig) (*Cache[T], error) {
 // getItem returns an item from the cache by height.
 // Returns nil if not found or type mismatch.
 func (c *Cache[T]) getItem(height uint64) *T {
-	c.itemsByHeightMu.RLock()
-	defer c.itemsByHeightMu.RUnlock()
-
 	item, ok := c.itemsByHeight.Get(height)
 	if !ok {
 		return nil
@@ -123,9 +122,6 @@ func (c *Cache[T]) getItem(height uint64) *T {
 
 // setItem sets an item in the cache by height
 func (c *Cache[T]) setItem(height uint64, item *T) {
-	c.itemsByHeightMu.Lock()
-	defer c.itemsByHeightMu.Unlock()
-
 	c.itemsByHeight.Add(height, item)
 }
 
@@ -145,9 +141,6 @@ func (c *Cache[T]) getNextItem(height uint64) *T {
 
 // isSeen returns true if the hash has been seen
 func (c *Cache[T]) isSeen(hash string) bool {
-	c.hashesMu.RLock()
-	defer c.hashesMu.RUnlock()
-
 	seen, ok := c.hashes.Get(hash)
 	if !ok {
 		return false
@@ -157,20 +150,12 @@ func (c *Cache[T]) isSeen(hash string) bool {
 
 // setSeen sets the hash as seen and tracks its height for pruning
 func (c *Cache[T]) setSeen(hash string, height uint64) {
-	c.hashesMu.Lock()
 	c.hashes.Add(hash, true)
-	c.hashesMu.Unlock()
-
-	c.hashByHeightMu.Lock()
 	c.hashByHeight.Add(height, hash)
-	c.hashByHeightMu.Unlock()
 }
 
 // getDAIncluded returns the DA height if the hash has been DA-included, otherwise it returns 0.
 func (c *Cache[T]) getDAIncluded(hash string) (uint64, bool) {
-	c.daIncludedMu.RLock()
-	defer c.daIncludedMu.RUnlock()
-
 	daHeight, ok := c.daIncluded.Get(hash)
 	if !ok {
 		return 0, false
@@ -180,13 +165,8 @@ func (c *Cache[T]) getDAIncluded(hash string) (uint64, bool) {
 
 // setDAIncluded sets the hash as DA-included with the given DA height and tracks block height for pruning
 func (c *Cache[T]) setDAIncluded(hash string, daHeight uint64, blockHeight uint64) {
-	c.daIncludedMu.Lock()
 	c.daIncluded.Add(hash, daHeight)
-	c.daIncludedMu.Unlock()
-
-	c.hashByHeightMu.Lock()
 	c.hashByHeight.Add(blockHeight, hash)
-	c.hashByHeightMu.Unlock()
 
 	// Update max DA height if necessary
 	for range 1_000 {
@@ -202,9 +182,6 @@ func (c *Cache[T]) setDAIncluded(hash string, daHeight uint64, blockHeight uint6
 
 // removeDAIncluded removes the DA-included status of the hash
 func (c *Cache[T]) removeDAIncluded(hash string) {
-	c.daIncludedMu.Lock()
-	defer c.daIncludedMu.Unlock()
-
 	c.daIncluded.Remove(hash)
 }
 
@@ -216,30 +193,24 @@ func (c *Cache[T]) daHeight() uint64 {
 
 // removeSeen removes a hash from the seen cache.
 func (c *Cache[T]) removeSeen(hash string) {
-	c.hashesMu.Lock()
-	defer c.hashesMu.Unlock()
 	c.hashes.Remove(hash)
 }
 
 // forEachHash iterates over all hashes in the seen cache and calls the provided function.
 // If the function returns false, iteration stops.
+// Note: iteration is best-effort; concurrent modifications may not be reflected.
 func (c *Cache[T]) forEachHash(fn func(hash string) bool) {
-	c.hashesMu.RLock()
 	keys := c.hashes.Keys()
-	c.hashesMu.RUnlock()
-
 	for _, hash := range keys {
 		if !fn(hash) {
 			break
 		}
 	}
 }
 
-// deleteAllForHeight removes all items and their associated data from the cache at the given height
+// deleteAllForHeight removes all items and their associated data from the cache at the given height.
 func (c *Cache[T]) deleteAllForHeight(height uint64) {
-	c.itemsByHeightMu.Lock()
 	c.itemsByHeight.Remove(height)
-	c.itemsByHeightMu.Unlock()
 
 	c.hashByHeightMu.Lock()
 	hash, ok := c.hashByHeight.Get(height)
@@ -249,9 +220,7 @@ func (c *Cache[T]) deleteAllForHeight(height uint64) {
 	c.hashByHeightMu.Unlock()
 
 	if ok {
-		c.hashesMu.Lock()
 		c.hashes.Remove(hash)
-		c.hashesMu.Unlock()
 		// c.daIncluded.Remove(hash) // we actually do not want to delete the DA-included status here
 	}
 }
@@ -307,62 +276,55 @@ func (c *Cache[T]) SaveToDisk(folderPath string) error {
 		return fmt.Errorf("failed to create directory %s: %w", folderPath, err)
 	}
 	var wg errgroup.Group
-	// prepare items maps
+
+	// save items by height
 	wg.Go(func() error {
 		itemsByHeightMap := make(map[uint64]*T)
-
-		c.itemsByHeightMu.RLock()
 		keys := c.itemsByHeight.Keys()
 		for _, k := range keys {
 			if v, ok := c.itemsByHeight.Peek(k); ok {
 				itemsByHeightMap[k] = v
 			}
 		}
-		c.itemsByHeightMu.RUnlock()
 
 		if err := saveMapGob(filepath.Join(folderPath, itemsByHeightFilename), itemsByHeightMap); err != nil {
 			return fmt.Errorf("save %s: %w", itemsByHeightFilename, err)
 		}
 		return nil
 	})
 
-	// prepare hashes map
+	// save hashes
 	wg.Go(func() error {
 		hashesToSave := make(map[string]bool)
-
-		c.hashesMu.RLock()
 		keys := c.hashes.Keys()
 		for _, k := range keys {
 			if v, ok := c.hashes.Peek(k); ok {
 				hashesToSave[k] = v
 			}
 		}
-		c.hashesMu.RUnlock()
 
 		if err := saveMapGob(filepath.Join(folderPath, hashesFilename), hashesToSave); err != nil {
 			return fmt.Errorf("save %s: %w", hashesFilename, err)
 		}
 		return nil
 	})
 
-	// prepare daIncluded map
+	// save daIncluded
 	wg.Go(func() error {
 		daIncludedToSave := make(map[string]uint64)
-
-		c.daIncludedMu.RLock()
 		keys := c.daIncluded.Keys()
 		for _, k := range keys {
 			if v, ok := c.daIncluded.Peek(k); ok {
 				daIncludedToSave[k] = v
 			}
 		}
-		c.daIncludedMu.RUnlock()
 
 		if err := saveMapGob(filepath.Join(folderPath, daIncludedFilename), daIncludedToSave); err != nil {
 			return fmt.Errorf("save %s: %w", daIncludedFilename, err)
 		}
 		return nil
 	})
+
 	return wg.Wait()
 }
 
@@ -372,39 +334,37 @@ func (c *Cache[T]) SaveToDisk(folderPath string) error {
 // are registered with the gob package if necessary (e.g., using gob.Register).
 func (c *Cache[T]) LoadFromDisk(folderPath string) error {
 	var wg errgroup.Group
+
 	// load items by height
 	wg.Go(func() error {
 		itemsByHeightMap, err := loadMapGob[uint64, *T](filepath.Join(folderPath, itemsByHeightFilename))
 		if err != nil {
 			return fmt.Errorf("failed to load %s : %w", itemsByHeightFilename, err)
 		}
-		c.itemsByHeightMu.Lock()
 		for k, v := range itemsByHeightMap {
 			c.itemsByHeight.Add(k, v)
 		}
-		c.itemsByHeightMu.Unlock()
 		return nil
 	})
+
 	// load hashes
 	wg.Go(func() error {
 		hashesMap, err := loadMapGob[string, bool](filepath.Join(folderPath, hashesFilename))
 		if err != nil {
 			return fmt.Errorf("failed to load %s : %w", hashesFilename, err)
 		}
-		c.hashesMu.Lock()
 		for k, v := range hashesMap {
 			c.hashes.Add(k, v)
 		}
-		c.hashesMu.Unlock()
 		return nil
 	})
+
 	// load daIncluded
 	wg.Go(func() error {
 		daIncludedMap, err := loadMapGob[string, uint64](filepath.Join(folderPath, daIncludedFilename))
 		if err != nil {
 			return fmt.Errorf("failed to load %s : %w", daIncludedFilename, err)
 		}
-		c.daIncludedMu.Lock()
 		for k, v := range daIncludedMap {
 			c.daIncluded.Add(k, v)
 			// Update max DA height during load
@@ -413,8 +373,8 @@ func (c *Cache[T]) LoadFromDisk(folderPath string) error {
 				c.maxDAHeight.Store(v)
 			}
 		}
-		c.daIncludedMu.Unlock()
 		return nil
 	})
+
 	return wg.Wait()
 }

pkg/store/cached_store.go

@@ -2,7 +2,6 @@ package store
 
 import (
 	"context"
-	"sync"
 
 	lru "github.com/hashicorp/golang-lru/v2"
 
@@ -20,15 +19,12 @@ const (
 )
 
 // CachedStore wraps a Store with LRU caching for frequently accessed data.
+// The underlying LRU cache is thread-safe, so no additional synchronization is needed.
 type CachedStore struct {
 	Store
 
-	headerCache   *lru.Cache[uint64, *types.SignedHeader]
-	headerCacheMu sync.RWMutex
-
-	// Optional: cache for block data (headers + data together)
-	blockDataCache   *lru.Cache[uint64, *blockDataEntry]
-	blockDataCacheMu sync.RWMutex
+	headerCache    *lru.Cache[uint64, *types.SignedHeader]
+	blockDataCache *lru.Cache[uint64, *blockDataEntry]
 }
 
 type blockDataEntry struct {
@@ -93,12 +89,9 @@ func NewCachedStore(store Store, opts ...CachedStoreOption) (*CachedStore, error
 // GetHeader returns the header at the given height, using the cache if available.
 func (cs *CachedStore) GetHeader(ctx context.Context, height uint64) (*types.SignedHeader, error) {
 	// Try cache first
-	cs.headerCacheMu.RLock()
 	if header, ok := cs.headerCache.Get(height); ok {
-		cs.headerCacheMu.RUnlock()
 		return header, nil
 	}
-	cs.headerCacheMu.RUnlock()
 
 	// Cache miss, fetch from underlying store
 	header, err := cs.Store.GetHeader(ctx, height)
@@ -107,22 +100,17 @@ func (cs *CachedStore) GetHeader(ctx context.Context, height uint64) (*types.Sig
 	}
 
 	// Add to cache
-	cs.headerCacheMu.Lock()
 	cs.headerCache.Add(height, header)
-	cs.headerCacheMu.Unlock()
 
 	return header, nil
 }
 
 // GetBlockData returns block header and data at given height, using cache if available.
 func (cs *CachedStore) GetBlockData(ctx context.Context, height uint64) (*types.SignedHeader, *types.Data, error) {
 	// Try cache first
-	cs.blockDataCacheMu.RLock()
 	if entry, ok := cs.blockDataCache.Get(height); ok {
-		cs.blockDataCacheMu.RUnlock()
 		return entry.header, entry.data, nil
 	}
-	cs.blockDataCacheMu.RUnlock()
 
 	// Cache miss, fetch from underlying store
 	header, data, err := cs.Store.GetBlockData(ctx, height)
@@ -131,42 +119,26 @@ func (cs *CachedStore) GetBlockData(ctx context.Context, height uint64) (*types.
 	}
 
 	// Add to cache
-	cs.blockDataCacheMu.Lock()
 	cs.blockDataCache.Add(height, &blockDataEntry{header: header, data: data})
-	cs.blockDataCacheMu.Unlock()
 
 	// Also add header to header cache
-	cs.headerCacheMu.Lock()
 	cs.headerCache.Add(height, header)
-	cs.headerCacheMu.Unlock()
 
 	return header, data, nil
 }
 
 // InvalidateRange removes headers in the given range from the cache.
 func (cs *CachedStore) InvalidateRange(fromHeight, toHeight uint64) {
-	cs.headerCacheMu.Lock()
 	for h := fromHeight; h <= toHeight; h++ {
 		cs.headerCache.Remove(h)
-	}
-	cs.headerCacheMu.Unlock()
-
-	cs.blockDataCacheMu.Lock()
-	for h := fromHeight; h <= toHeight; h++ {
 		cs.blockDataCache.Remove(h)
 	}
-	cs.blockDataCacheMu.Unlock()
 }
 
 // ClearCache clears all cached entries.
 func (cs *CachedStore) ClearCache() {
-	cs.headerCacheMu.Lock()
 	cs.headerCache.Purge()
-	cs.headerCacheMu.Unlock()
-
-	cs.blockDataCacheMu.Lock()
 	cs.blockDataCache.Purge()
-	cs.blockDataCacheMu.Unlock()
 }
 
 // Rollback wraps the underlying store's Rollback and invalidates affected cache entries.