Algod: Add a new caching mechanism for the ledger - latest 512 block headers

ledger/blockHeaderCache.go

@@ -0,0 +1,86 @@
+// Copyright (C) 2019-2022 Algorand, Inc.
+// This file is part of go-algorand
+//
+// go-algorand is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as
+// published by the Free Software Foundation, either version 3 of the
+// License, or (at your option) any later version.
+//
+// go-algorand is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with go-algorand.  If not, see <https://www.gnu.org/licenses/>.
+
+package ledger
+
+import (
+	"github.com/algorand/go-algorand/data/basics"
+	"github.com/algorand/go-algorand/data/bookkeeping"
+
+	"github.com/algorand/go-deadlock"
+)
+
+const latestHeaderCacheSize = 512
+const blockHeadersLRUCacheSize = 10
+
+// blockHeaderCache is a wrapper for all block header cache mechanisms used within the Ledger.
+type blockHeaderCache struct {
+	lruCache          heapLRUCache
+	latestHeaderCache latestBlockHeaderCache
+}
+
+type latestBlockHeaderCache struct {
+	blockHeaders [latestHeaderCacheSize]bookkeeping.BlockHeader
+	mutex        deadlock.RWMutex
+}
+
+func (c *blockHeaderCache) initialize() {
+	c.lruCache.maxEntries = blockHeadersLRUCacheSize
+}
+
+func (c *blockHeaderCache) get(round basics.Round) (blockHeader bookkeeping.BlockHeader, exists bool) {
+	// check latestHeaderCache first
+	blockHeader, exists = c.latestHeaderCache.get(round)
+	if exists {
+		return
+	}
+
+	// if not found in latestHeaderCache, check LRUCache
+	value, exists := c.lruCache.Get(round)
+	if exists {
+		blockHeader = value.(bookkeeping.BlockHeader)
+	}
+
+	return
+}
+
+func (c *blockHeaderCache) put(blockHeader bookkeeping.BlockHeader) {
+	c.latestHeaderCache.put(blockHeader)
+	c.lruCache.Put(blockHeader.Round, blockHeader)
+}
+
+func (c *latestBlockHeaderCache) get(round basics.Round) (blockHeader bookkeeping.BlockHeader, exists bool) {
+	c.mutex.RLock()
+	defer c.mutex.RUnlock()
+
+	idx := round % latestHeaderCacheSize
+	if round == 0 || c.blockHeaders[idx].Round != round { // blockHeader is empty or not requested round
+		return bookkeeping.BlockHeader{}, false
+	}
+	blockHeader = c.blockHeaders[idx]
+
+	return blockHeader, true
+}
+
+func (c *latestBlockHeaderCache) put(blockHeader bookkeeping.BlockHeader) {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	idx := blockHeader.Round % latestHeaderCacheSize
+	if blockHeader.Round > c.blockHeaders[idx].Round { // provided blockHeader is more recent than cached one
+		c.blockHeaders[idx] = blockHeader
+	}
+}

ledger/blockHeaderCache_test.go

@@ -0,0 +1,95 @@
+// Copyright (C) 2019-2022 Algorand, Inc.
+// This file is part of go-algorand
+//
+// go-algorand is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as
+// published by the Free Software Foundation, either version 3 of the
+// License, or (at your option) any later version.
+//
+// go-algorand is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with go-algorand.  If not, see <https://www.gnu.org/licenses/>.
+
+package ledger
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/algorand/go-algorand/config"
+	"github.com/algorand/go-algorand/data/basics"
+	"github.com/algorand/go-algorand/data/bookkeeping"
+	"github.com/algorand/go-algorand/protocol"
+	"github.com/algorand/go-algorand/test/partitiontest"
+)
+
+func TestBlockHeaderCache(t *testing.T) {
+	partitiontest.PartitionTest(t)
+	a := require.New(t)
+
+	var cache blockHeaderCache
+	cache.initialize()
+	for i := basics.Round(1024); i < 1024+latestHeaderCacheSize; i++ {
+		hdr := bookkeeping.BlockHeader{Round: i}
+		cache.put(hdr)
+	}
+
+	rnd := basics.Round(120)
+	hdr := bookkeeping.BlockHeader{Round: rnd}
+	cache.put(hdr)
+
+	_, exists := cache.get(rnd)
+	a.True(exists)
+
+	_, exists = cache.lruCache.Get(rnd)
+	a.True(exists)
+
+	_, exists = cache.latestHeaderCache.get(rnd)
+	a.False(exists)
+
+	rnd = basics.Round(2048)
+	hdr = bookkeeping.BlockHeader{Round: rnd}
+	cache.put(hdr)
+
+	_, exists = cache.latestHeaderCache.get(rnd)
+	a.True(exists)
+
+	_, exists = cache.lruCache.Get(rnd)
+	a.True(exists)
+
+}
+
+func TestLatestBlockHeaderCache(t *testing.T) {
+	partitiontest.PartitionTest(t)
+	a := require.New(t)
+
+	var cache latestBlockHeaderCache
+	for i := basics.Round(123); i < latestHeaderCacheSize; i++ {
+		hdr := bookkeeping.BlockHeader{Round: i}
+		cache.put(hdr)
+	}
+
+	for i := basics.Round(0); i < 123; i++ {
+		_, exists := cache.get(i)
+		a.False(exists)
+	}
+
+	for i := basics.Round(123); i < latestHeaderCacheSize; i++ {
+		hdr, exists := cache.get(i)
+		a.True(exists)
+		a.Equal(i, hdr.Round)
+	}
+}
+
+func TestCacheSizeConsensus(t *testing.T) {
+	partitiontest.PartitionTest(t)
+	a := require.New(t)
+
+	// TODO Stateproof: change to CurrentVersion when feature is enabled
+	a.Equal(uint64(latestHeaderCacheSize), config.Consensus[protocol.ConsensusFuture].CompactCertRounds*2)
+}

ledger/ledger.go

@@ -85,7 +85,7 @@ type Ledger struct {
 	trackers  trackerRegistry
 	trackerMu deadlock.RWMutex
 
-	headerCache heapLRUCache
+	headerCache blockHeaderCache
 
 	// verifiedTxnCache holds all the verified transactions state
 	verifiedTxnCache verify.VerifiedTransactionCache
@@ -119,7 +119,7 @@ func OpenLedger(
 		cfg:                            cfg,
 	}
 
-	l.headerCache.maxEntries = 10
+	l.headerCache.initialize()
 
 	defer func() {
 		if err != nil {
@@ -583,15 +583,14 @@ func (l *Ledger) Block(rnd basics.Round) (blk bookkeeping.Block, err error) {
 
 // BlockHdr returns the BlockHeader of the block for round rnd.
 func (l *Ledger) BlockHdr(rnd basics.Round) (blk bookkeeping.BlockHeader, err error) {
-	value, exists := l.headerCache.Get(rnd)
+	blk, exists := l.headerCache.get(rnd)
 	if exists {
-		blk = value.(bookkeeping.BlockHeader)
 		return
 	}
 
 	blk, err = l.blockQ.getBlockHdr(rnd)
 	if err == nil {
-		l.headerCache.Put(rnd, blk)
+		l.headerCache.put(blk)
 	}
 	return
 }
@@ -641,7 +640,7 @@ func (l *Ledger) AddValidatedBlock(vb ledgercore.ValidatedBlock, cert agreement.
 	if err != nil {
 		return err
 	}
-	l.headerCache.Put(blk.Round(), blk.BlockHeader)
+	l.headerCache.put(blk.BlockHeader)
 	l.trackers.newBlock(blk, vb.Delta())
 	l.log.Debugf("ledger.AddValidatedBlock: added blk %d", blk.Round())
 	return nil

ledger/ledger_test.go

@@ -1363,6 +1363,7 @@ func testLedgerRegressionFaultyLeaseFirstValidCheck2f3880f7(t *testing.T, versio
 
 func TestLedgerBlockHdrCaching(t *testing.T) {
 	partitiontest.PartitionTest(t)
+	a := require.New(t)
 
 	dbName := fmt.Sprintf("%s.%d", t.Name(), crypto.RandUint64())
 	genesisInitState := getInitState()
@@ -1371,20 +1372,89 @@ func TestLedgerBlockHdrCaching(t *testing.T) {
 	cfg.Archival = true
 	log := logging.TestingLog(t)
 	l, err := OpenLedger(log, dbName, inMem, genesisInitState, cfg)
-	require.NoError(t, err)
+	a.NoError(err)
 	defer l.Close()
 
 	blk := genesisInitState.Block
 
-	for i := 0; i < 128; i++ {
+	for i := 0; i < 1024; i++ {
 		blk.BlockHeader.Round++
 		blk.BlockHeader.TimeStamp += int64(crypto.RandUint64() % 100 * 1000)
 		err := l.AddBlock(blk, agreement.Certificate{})
-		require.NoError(t, err)
+		a.NoError(err)
 
 		hdr, err := l.BlockHdr(blk.BlockHeader.Round)
-		require.NoError(t, err)
-		require.Equal(t, blk.BlockHeader, hdr)
+		a.NoError(err)
+		a.Equal(blk.BlockHeader, hdr)
+	}
+
+	rnd := basics.Round(128)
+	hdr, err := l.BlockHdr(rnd) // should update LRU cache but not latestBlockHeaderCache
+	a.NoError(err)
+	a.Equal(rnd, hdr.Round)
+
+	_, exists := l.headerCache.lruCache.Get(rnd)
+	a.True(exists)
+
+	_, exists = l.headerCache.latestHeaderCache.get(rnd)
+	a.False(exists)
+}
+
+func BenchmarkLedgerBlockHdrCaching(b *testing.B) {
+	benchLedgerCache(b, 1024-256+1)
+}
+
+func BenchmarkLedgerBlockHdrWithoutCaching(b *testing.B) {
+	benchLedgerCache(b, 100)
+}
+
+type nullWriter struct{} // logging output not required
+
+func (w nullWriter) Write(data []byte) (n int, err error) {
+	return len(data), nil
+}
+
+func benchLedgerCache(b *testing.B, startRound basics.Round) {
+	a := require.New(b)
+
+	dbName := fmt.Sprintf("%s.%d", b.Name(), crypto.RandUint64())
+	genesisInitState := getInitState()
+	const inMem = false // benchmark actual DB stored in disk instead of on memory
+	cfg := config.GetDefaultLocal()
+	cfg.Archival = true
+	log := logging.TestingLog(b)
+	log.SetOutput(nullWriter{})
+	l, err := OpenLedger(log, dbName, inMem, genesisInitState, cfg)
+	a.NoError(err)
+	defer func() { // close ledger and remove temporary DB file
+		l.Close()
+		err := os.Remove(dbName + ".tracker.sqlite")
+		if err != nil {
+			fmt.Printf("os.Remove: %v \n", err)
+		}
+		err = os.Remove(dbName + ".block.sqlite")
+		if err != nil {
+			fmt.Printf("os.Remove: %v \n", err)
+		}
+
+	}()
+
+	blk := genesisInitState.Block
+
+	// Fill ledger (and its cache) with blocks
+	for i := 0; i < 1024; i++ {
+		blk.BlockHeader.Round++
+		blk.BlockHeader.TimeStamp += int64(crypto.RandUint64() % 100 * 1000)
+		err := l.AddBlock(blk, agreement.Certificate{})
+		a.NoError(err)
+	}
+
+	for i := 0; i < b.N; i++ {
+		for j := startRound; j < startRound+256; j++ { // these rounds should be in cache
+			hdr, err := l.BlockHdr(j)
+			a.NoError(err)
+			a.Equal(j, hdr.Round)
+		}
 	}
 }