core/rawdb/eradb: improve safety of store

This removes the use of atomic reference counters in favor of using a scheme where all
changes to reference counters are done while holding the Store mutex. Store.Close now
synchronizes with all running accesses and waits until all files have been closed
before returning.

core/rawdb/eradb/eradb.go

@@ -14,6 +14,7 @@
 // You should have received a copy of the GNU Lesser General Public License
 // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
 
+// Package eradb implements a history backend using era1 files.
 package eradb
 
 import (
@@ -23,7 +24,6 @@ import (
 	"os"
 	"path/filepath"
 	"sync"
-	"sync/atomic"
 
 	"github.com/ethereum/go-ethereum/common/lru"
 	"github.com/ethereum/go-ethereum/internal/era"
@@ -34,20 +34,33 @@ const openFileLimit = 64
 
 var errClosed = errors.New("era store is closed")
 
+type fileCacheStat byte
+
+const (
+	storeClosing fileCacheStat = iota
+	fileIsNew
+	fileIsOpening
+	fileIsCached
+)
+
 // Store manages read access to a directory of era1 files.
 // The getter methods are thread-safe.
 type Store struct {
 	datadir string
+
+	// The mutex protects all remaining fields.
 	mu      sync.Mutex
+	cond    *sync.Cond
 	lru     lru.BasicLRU[uint64, *fileCacheEntry]
 	opening map[uint64]*fileCacheEntry
+	closing bool
 }
 
 type fileCacheEntry struct {
-	ref    atomic.Int32
-	opened chan struct{}
-	file   *era.Era
-	err    error
+	refcount int           // reference count. This is protected by Store.mu!
+	opened   chan struct{} // signals opening of file has completed
+	file     *era.Era      // the file
+	err      error         // error from opening the file
 }
 
 // New opens the store directory.
@@ -68,6 +81,7 @@ func New(datadir string) (*Store, error) {
 		lru:     lru.NewBasicLRU[uint64, *fileCacheEntry](openFileLimit),
 		opening: make(map[uint64]*fileCacheEntry),
 	}
+	db.cond = sync.NewCond(&db.mu)
 	log.Info("Opened Era store", "datadir", datadir)
 	return db, nil
 }
@@ -77,12 +91,23 @@ func (db *Store) Close() {
 	db.mu.Lock()
 	defer db.mu.Unlock()
 
-	keys := db.lru.Keys()
-	for _, epoch := range keys {
-		entry, _ := db.lru.Get(epoch)
-		entry.done(epoch)
+	// Prevent new cache additions.
+	db.closing = true
+
+	// Deref all active files. Since inactive files have a refcount of one, they will be
+	// closed right here and now after decrementing. Files which are currently being used
+	// have a refcount > 1 and will hit zero when their access finishes.
+	for _, epoch := range db.lru.Keys() {
+		entry, _ := db.lru.Peek(epoch)
+		if entry.derefAndClose(epoch) {
+			db.lru.Remove(epoch)
+		}
+	}
+
+	// Wait for all store access to finish.
+	for db.lru.Len() > 0 || len(db.opening) > 0 {
+		db.cond.Wait()
 	}
-	db.opening = nil
 }
 
 // GetRawBody returns the raw body for a given block number.
@@ -95,7 +120,8 @@ func (db *Store) GetRawBody(number uint64) ([]byte, error) {
 		}
 		return nil, entry.err
 	}
-	defer entry.done(epoch)
+	defer db.doneWithFile(epoch, entry)
+
 	return entry.file.GetRawBodyByNumber(number)
 }
 
@@ -109,21 +135,24 @@ func (db *Store) GetRawReceipts(number uint64) ([]byte, error) {
 		}
 		return nil, entry.err
 	}
-	defer entry.done(epoch)
+	defer db.doneWithFile(epoch, entry)
+
 	return entry.file.GetRawReceiptsByNumber(number)
 }
 
-// getEraByEpoch opens an era file or gets it from the cache. The caller can access
-// entry.file and entry.err and must call entry.done when done reading the file.
+// getEraByEpoch opens an era file or gets it from the cache.
+// The caller can freely access the returned entry's .file and .err
+// db.doneWithFile must be called when it is done reading the file.
 func (db *Store) getEraByEpoch(epoch uint64) *fileCacheEntry {
 	// Add the requested epoch to the cache.
-	entry := db.getCacheEntry(epoch)
-	if entry == nil {
-		return &fileCacheEntry{err: errClosed}
-	}
+	stat, entry := db.getCacheEntry(epoch)
 
-	// First goroutine to use the file has to open it.
-	if entry.ref.Add(1) == 1 {
+	switch stat {
+	case storeClosing:
+		return &fileCacheEntry{err: errClosed}
+
+	case fileIsNew:
+		// Open the file and put it into the cache.
 		e, err := db.openEraFile(epoch)
 		if err != nil {
 			db.fileFailedToOpen(epoch, entry, err)
@@ -131,34 +160,44 @@ func (db *Store) getEraByEpoch(epoch uint64) *fileCacheEntry {
 			db.fileOpened(epoch, entry, e)
 		}
 		close(entry.opened)
-	}
 
-	// Bump the refcount and wait for the file to be opened.
-	entry.ref.Add(1)
-	<-entry.opened
+	case fileIsOpening:
+		// Wait for open to finish.
+		<-entry.opened
+
+	case fileIsCached:
+		// Nothing to do.
+
+	default:
+		panic(fmt.Sprintf("invalid file state %d", stat))
+	}
 	return entry
 }
 
 // getCacheEntry gets an open era file from the cache.
-func (db *Store) getCacheEntry(epoch uint64) *fileCacheEntry {
+func (db *Store) getCacheEntry(epoch uint64) (stat fileCacheStat, entry *fileCacheEntry) {
 	db.mu.Lock()
 	defer db.mu.Unlock()
 
-	// Check if this epoch is already being opened.
-	if db.opening == nil {
-		return nil
+	if db.closing {
+		return storeClosing, nil
 	}
-	if entry, ok := db.opening[epoch]; ok {
-		return entry
+	if entry = db.opening[epoch]; entry != nil {
+		stat = fileIsOpening
+	} else if entry, _ = db.lru.Get(epoch); entry != nil {
+		stat = fileIsCached
+	} else {
+		// It's a new file, create an entry in the opening table. Note the entry is
+		// created with an initial refcount of one. We increment the count once more
+		// before returning, but the count will return to one when the file has been
+		// accessed. When the store is closed or the file gets evicted from the cache,
+		// refcount will be decreased by one, thus allowing it to hit zero.
+		entry = &fileCacheEntry{refcount: 1, opened: make(chan struct{})}
+		db.opening[epoch] = entry
+		stat = fileIsNew
 	}
-	// Check if it's in the cache.
-	if entry, ok := db.lru.Get(epoch); ok {
-		return entry
-	}
-	// It's a new file, create an entry in the 'opening' table.
-	entry := &fileCacheEntry{opened: make(chan struct{})}
-	db.opening[epoch] = entry
-	return entry
+	entry.refcount++
+	return stat, entry
 }
 
 // fileOpened is called after an era file has been successfully opened.
@@ -166,32 +205,33 @@ func (db *Store) fileOpened(epoch uint64, entry *fileCacheEntry, file *era.Era)
 	db.mu.Lock()
 	defer db.mu.Unlock()
 
-	// The database may have been closed while opening the file. When that happens,
-	// db.opening will be set to nil, so we need to handle that here and ensure the caller
-	// knows.
-	if db.opening == nil {
+	delete(db.opening, epoch)
+	db.cond.Signal() // db.opening was modified
+
+	// The database may have been closed while opening the file. When that happens, we
+	// need to close the file here, since it isn't tracked by the LRU yet.
+	if db.closing {
 		entry.err = errClosed
+		file.Close()
 		return
 	}
 
-	// Remove from 'opening' table and add to the LRU.
-	// This may evict an existing item, which we have to close.
+	// Add it to the LRU. This may evict an existing item, which we have to close.
 	entry.file = file
-	delete(db.opening, epoch)
-	if _, evictedEntry, _ := db.lru.Add3(epoch, entry); evictedEntry != nil {
-		evictedEntry.done(epoch)
+	evictedEpoch, evictedEntry, _ := db.lru.Add3(epoch, entry)
+	if evictedEntry != nil {
+		evictedEntry.derefAndClose(evictedEpoch)
 	}
 }
 
 // fileFailedToOpen is called when an era file could not be opened.
 func (db *Store) fileFailedToOpen(epoch uint64, entry *fileCacheEntry, err error) {
-	entry.err = err
-
 	db.mu.Lock()
 	defer db.mu.Unlock()
-	if db.opening != nil {
-		delete(db.opening, epoch)
-	}
+
+	delete(db.opening, epoch)
+	db.cond.Signal() // db.opening was modified
+	entry.err = err
 }
 
 func (db *Store) openEraFile(epoch uint64) (*era.Era, error) {
@@ -223,18 +263,35 @@ func (db *Store) openEraFile(epoch uint64) (*era.Era, error) {
 	return e, nil
 }
 
-// done signals that the caller has finished using a file.
+// doneWithFile signals that the caller has finished using a file.
 // This decrements the refcount and ensures the file is closed by the last user.
-func (f *fileCacheEntry) done(epoch uint64) {
-	if f.err != nil {
+func (db *Store) doneWithFile(epoch uint64, entry *fileCacheEntry) {
+	db.mu.Lock()
+	defer db.mu.Unlock()
+
+	if entry.err != nil {
 		return
 	}
-	if f.ref.Add(-1) == 0 {
-		err := f.file.Close()
-		if err == nil {
-			log.Debug("Closed era1 file", "epoch", epoch)
-		} else {
-			log.Warn("Error closing era1 file", "epoch", epoch, "err", err)
+	if entry.derefAndClose(epoch) {
+		// Delete closed entry from LRU if it is still present.
+		if e, _ := db.lru.Peek(epoch); e == entry {
+			db.lru.Remove(epoch)
+			db.cond.Signal() // db.lru was modified
 		}
 	}
 }
+
+func (entry *fileCacheEntry) derefAndClose(epoch uint64) (closed bool) {
+	entry.refcount--
+	if entry.refcount > 0 {
+		return false
+	}
+
+	closeErr := entry.file.Close()
+	if closeErr == nil {
+		log.Debug("Closed era1 file", "epoch", epoch)
+	} else {
+		log.Warn("Error closing era1 file", "epoch", epoch, "err", closeErr)
+	}
+	return true
+}