Repository: kaspanet/kaspad
Pull request #1334 (2020-12-31 14:01:31)

Remove LevelDBSnapshot from LevelDBTransaction
We currently don't use the snapshot in the transaction, in fact we actually do the opposite logic in our Stores (check against the staging + against toDelete staging)

this will hopefully improve some of the DB performance


Patch:
diff --git a/app/protocol/flowcontext/blocks.go b/app/protocol/flowcontext/blocks.go
index 1ab87e81..96965d24 100644
--- a/app/protocol/flowcontext/blocks.go
+++ b/app/protocol/flowcontext/blocks.go
@@ -24,7 +24,7 @@ func (f *FlowContext) OnNewBlock(block *externalapi.DomainBlock,
 	log.Debugf("OnNewBlock start for block %s", hash)
 	defer log.Debugf("OnNewBlock end for block %s", hash)
 
-	f.updateRecentBlockAddedTimesWithLastBlock()
+	f.UpdateRecentBlockAddedTimesWithLastBlock()
 	unorphaningResults, err := f.UnorphanBlocks(block)
 	if err != nil {
 		return err
diff --git a/app/protocol/flowcontext/orphans.go b/app/protocol/flowcontext/orphans.go
index f655c4bc..d7a17ab5 100644
--- a/app/protocol/flowcontext/orphans.go
+++ b/app/protocol/flowcontext/orphans.go
@@ -156,7 +156,7 @@ func (f *FlowContext) unorphanBlock(orphanHash externalapi.DomainHash) (*externa
 		}
 		return nil, false, err
 	}
-	f.updateRecentBlockAddedTimesWithLastBlock()
+	f.UpdateRecentBlockAddedTimesWithLastBlock()
 
 	log.Infof("Unorphaned block %s", orphanHash)
 	return blockInsertionResult, true, nil
diff --git a/app/protocol/flowcontext/sync_rate.go b/app/protocol/flowcontext/sync_rate.go
index b7c0f2d2..fedbc27e 100644
--- a/app/protocol/flowcontext/sync_rate.go
+++ b/app/protocol/flowcontext/sync_rate.go
@@ -3,12 +3,14 @@ package flowcontext
 import "github.com/kaspanet/kaspad/util/mstime"
 
 const (
-	syncRateWindowInMilliSeconds                         = 60_000
+	syncRateWindowInMilliSeconds                         = 15 * 60 * 1000
 	syncRateMaxDeviation                                 = 0.05
 	maxSelectedParentTimeDiffToAllowMiningInMilliSeconds = 300_000
 )
 
-func (f *FlowContext) updateRecentBlockAddedTimesWithLastBlock() {
+// UpdateRecentBlockAddedTimesWithLastBlock adds current time to list of times when block was added.
+// We use this list to determine the current sync rate
+func (f *FlowContext) UpdateRecentBlockAddedTimesWithLastBlock() {
 	f.recentBlockAddedTimesMutex.Lock()
 	defer f.recentBlockAddedTimesMutex.Unlock()
 
@@ -44,7 +46,14 @@ func (f *FlowContext) isSyncRateBelowMinimum() bool {
 	}
 
 	expectedBlocks := float64(syncRateWindowInMilliSeconds) / float64(f.cfg.NetParams().TargetTimePerBlock.Milliseconds())
-	return 1-float64(len(f.recentBlockAddedTimes))/expectedBlocks > syncRateMaxDeviation
+	isSyncRateTooLow := 1-float64(len(f.recentBlockAddedTimes))/expectedBlocks > syncRateMaxDeviation
+
+	if isSyncRateTooLow {
+		log.Debugf("In the last %d seconds, got %d blocks, while at least %f were expected.",
+			syncRateWindowInMilliSeconds/1000, len(f.recentBlockAddedTimes), expectedBlocks*(1-syncRateMaxDeviation))
+	}
+
+	return isSyncRateTooLow
 }
 
 // ShouldMine returns whether it's ok to use block template from this node
diff --git a/app/protocol/flows/blockrelay/handle_relay_invs.go b/app/protocol/flows/blockrelay/handle_relay_invs.go
index 8b91c578..f4a81a52 100644
--- a/app/protocol/flows/blockrelay/handle_relay_invs.go
+++ b/app/protocol/flows/blockrelay/handle_relay_invs.go
@@ -33,6 +33,7 @@ type RelayInvsContext interface {
 	IsIBDRunning() bool
 	TrySetIBDRunning() bool
 	UnsetIBDRunning()
+	UpdateRecentBlockAddedTimesWithLastBlock()
 }
 
 type handleRelayInvsFlow struct {
diff --git a/app/protocol/flows/blockrelay/ibd.go b/app/protocol/flows/blockrelay/ibd.go
index 9d70cc3c..0e266d0e 100644
--- a/app/protocol/flows/blockrelay/ibd.go
+++ b/app/protocol/flows/blockrelay/ibd.go
@@ -1,6 +1,8 @@
 package blockrelay
 
 import (
+	"time"
+
 	"github.com/kaspanet/kaspad/app/appmessage"
 	"github.com/kaspanet/kaspad/app/protocol/common"
 	"github.com/kaspanet/kaspad/app/protocol/protocolerrors"
@@ -8,7 +10,6 @@ import (
 	"github.com/kaspanet/kaspad/domain/consensus/ruleerrors"
 	"github.com/kaspanet/kaspad/domain/consensus/utils/consensushashing"
 	"github.com/pkg/errors"
-	"time"
 )
 
 func (flow *handleRelayInvsFlow) runIBDIfNotRunning(highHash *externalapi.DomainHash) error {
@@ -257,6 +258,9 @@ func (flow *handleRelayInvsFlow) processHeader(msgBlockHeader *appmessage.MsgBlo
 
 		return protocolerrors.Wrapf(true, err, "got invalid block %s during IBD", blockHash)
 	}
+
+	flow.UpdateRecentBlockAddedTimesWithLastBlock()
+
 	return nil
 }
 
diff --git a/domain/consensus/factory.go b/domain/consensus/factory.go
index 27828c46..37bef61f 100644
--- a/domain/consensus/factory.go
+++ b/domain/consensus/factory.go
@@ -82,7 +82,7 @@ func (f *factory) NewConsensus(dagParams *dagconfig.Params, db infrastructuredat
 	pruningStore := pruningstore.New()
 	reachabilityDataStore := reachabilitydatastore.New(pruningWindowSizeForCaches)
 	utxoDiffStore := utxodiffstore.New(200)
-	consensusStateStore := consensusstatestore.New()
+	consensusStateStore := consensusstatestore.New(10_000)
 	ghostdagDataStore := ghostdagdatastore.New(pruningWindowSizeForCaches)
 	headersSelectedTipStore := headersselectedtipstore.New()
 	finalityStore := finalitystore.New(200)
diff --git a/domain/consensus/utils/utxolrucache/utxolrucache.go b/domain/consensus/utils/utxolrucache/utxolrucache.go
new file mode 100644
index 00000000..70a9e879
--- /dev/null
+++ b/domain/consensus/utils/utxolrucache/utxolrucache.go
@@ -0,0 +1,71 @@
+package utxolrucache
+
+import (
+	"github.com/kaspanet/kaspad/domain/consensus/model/externalapi"
+)
+
+// LRUCache is a least-recently-used cache for UTXO entries
+// indexed by DomainOutpoint
+type LRUCache struct {
+	cache    map[externalapi.DomainOutpoint]externalapi.UTXOEntry
+	capacity int
+}
+
+// New creates a new LRUCache
+func New(capacity int) *LRUCache {
+	return &LRUCache{
+		cache:    make(map[externalapi.DomainOutpoint]externalapi.UTXOEntry, capacity+1),
+		capacity: capacity,
+	}
+}
+
+// Add adds an entry to the LRUCache
+func (c *LRUCache) Add(key *externalapi.DomainOutpoint, value externalapi.UTXOEntry) {
+	c.cache[*key] = value
+
+	if len(c.cache) > c.capacity {
+		c.evictRandom()
+	}
+}
+
+// Get returns the entry for the given key, or (nil, false) otherwise
+func (c *LRUCache) Get(key *externalapi.DomainOutpoint) (externalapi.UTXOEntry, bool) {
+	value, ok := c.cache[*key]
+	if !ok {
+		return nil, false
+	}
+	return value, true
+}
+
+// Has returns whether the LRUCache contains the given key
+func (c *LRUCache) Has(key *externalapi.DomainOutpoint) bool {
+	_, ok := c.cache[*key]
+	return ok
+}
+
+// Remove removes the entry for the the given key. Does nothing if
+// the entry does not exist
+func (c *LRUCache) Remove(key *externalapi.DomainOutpoint) {
+	delete(c.cache, *key)
+}
+
+// Clear clears the cache
+func (c *LRUCache) Clear() {
+	keys := make([]externalapi.DomainOutpoint, len(c.cache))
+	for outpoint := range c.cache {
+		keys = append(keys, outpoint)
+	}
+
+	for _, key := range keys {
+		delete(c.cache, key)
+	}
+}
+
+func (c *LRUCache) evictRandom() {
+	var keyToEvict externalapi.DomainOutpoint
+	for key := range c.cache {
+		keyToEvict = key
+		break
+	}
+	c.Remove(&keyToEvict)
+}
diff --git a/infrastructure/db/database/ldb/transaction.go b/infrastructure/db/database/ldb/transaction.go
index 5271ed71..095bc37f 100644
--- a/infrastructure/db/database/ldb/transaction.go
+++ b/infrastructure/db/database/ldb/transaction.go
@@ -8,37 +8,25 @@ import (
 )
 
 // LevelDBTransaction is a thin wrapper around native leveldb
-// batches and snapshots. It supports both get and put.
+// batches. It supports both get and put.
 //
-// Snapshots provide a frozen view of the database at the moment
-// the transaction begins. On the other hand, batches provide a
-// mechanism to combine several database writes into one write,
-// which seamlessly rolls back the database in case any individual
-// write fails. Together the two forms a logic unit similar
-// to what one might expect from a classic database transaction.
+// Note that reads are done from the Database directly, so if another transaction changed the data,
+// you will read the new data, and not the one from the time the transaction was opened/
 //
-// Note: Transactions provide data consistency over the state of
-// the database as it was when the transaction started. As it's
-// currently implemented, if one puts data into the transaction
+// Note: As it's currently implemented, if one puts data into the transaction
 // then it will not be available to get within the same transaction.
 type LevelDBTransaction struct {
 	db       *LevelDB
-	snapshot *leveldb.Snapshot
 	batch    *leveldb.Batch
 	isClosed bool
 }
 
 // Begin begins a new transaction.
 func (db *LevelDB) Begin() (database.Transaction, error) {
-	snapshot, err := db.ldb.GetSnapshot()
-	if err != nil {
-		return nil, errors.WithStack(err)
-	}
 	batch := new(leveldb.Batch)
 
 	transaction := &LevelDBTransaction{
 		db:       db,
-		snapshot: snapshot,
 		batch:    batch,
 		isClosed: false,
 	}
@@ -53,7 +41,6 @@ func (tx *LevelDBTransaction) Commit() error {
 	}
 
 	tx.isClosed = true
-	tx.snapshot.Release()
 	return errors.WithStack(tx.db.ldb.Write(tx.batch, &opt.WriteOptions{Sync: true}))
 }
 
@@ -65,7 +52,6 @@ func (tx *LevelDBTransaction) Rollback() error {
 	}
 
 	tx.isClosed = true
-	tx.snapshot.Release()
 	tx.batch.Reset()
 	return nil
 }
@@ -97,16 +83,7 @@ func (tx *LevelDBTransaction) Get(key *database.Key) ([]byte, error) {
 	if tx.isClosed {
 		return nil, errors.New("cannot get from a closed transaction")
 	}
-
-	data, err := tx.snapshot.Get(key.Bytes(), nil)
-	if err != nil {
-		if errors.Is(err, leveldb.ErrNotFound) {
-			return nil, errors.Wrapf(database.ErrNotFound,
-				"key %s not found", key)
-		}
-		return nil, errors.WithStack(err)
-	}
-	return data, nil
+	return tx.db.Get(key)
 }
 
 // Has returns true if the database does contains the
@@ -115,9 +92,7 @@ func (tx *LevelDBTransaction) Has(key *database.Key) (bool, error) {
 	if tx.isClosed {
 		return false, errors.New("cannot has from a closed transaction")
 	}
-
-	res, err := tx.snapshot.Has(key.Bytes(), nil)
-	return res, errors.WithStack(err)
+	return tx.db.Has(key)
 }
 
 // Delete deletes the value for the given key. Will not


Test patch:
diff --git a/domain/consensus/datastructures/consensusstatestore/consensusstatestore.go b/domain/consensus/datastructures/consensusstatestore/consensusstatestore.go
index b3a37a78..cc2d14b8 100644
--- a/domain/consensus/datastructures/consensusstatestore/consensusstatestore.go
+++ b/domain/consensus/datastructures/consensusstatestore/consensusstatestore.go
@@ -3,6 +3,7 @@ package consensusstatestore
 import (
 	"github.com/kaspanet/kaspad/domain/consensus/model"
 	"github.com/kaspanet/kaspad/domain/consensus/model/externalapi"
+	"github.com/kaspanet/kaspad/domain/consensus/utils/utxolrucache"
 )
 
 // consensusStateStore represents a store for the current consensus state
@@ -12,13 +13,17 @@ type consensusStateStore struct {
 	virtualUTXODiffStaging    model.UTXODiff
 	virtualUTXOSetStaging     model.UTXOCollection
 
+	virtualUTXOSetCache *utxolrucache.LRUCache
+
 	tipsCache               []*externalapi.DomainHash
 	virtualDiffParentsCache []*externalapi.DomainHash
 }
 
 // New instantiates a new ConsensusStateStore
-func New() model.ConsensusStateStore {
-	return &consensusStateStore{}
+func New(utxoSetCacheSize int) model.ConsensusStateStore {
+	return &consensusStateStore{
+		virtualUTXOSetCache: utxolrucache.New(utxoSetCacheSize),
+	}
 }
 
 func (css *consensusStateStore) Discard() {
diff --git a/domain/consensus/datastructures/consensusstatestore/utxo.go b/domain/consensus/datastructures/consensusstatestore/utxo.go
index 6bc76113..5959bfb1 100644
--- a/domain/consensus/datastructures/consensusstatestore/utxo.go
+++ b/domain/consensus/datastructures/consensusstatestore/utxo.go
@@ -39,6 +39,9 @@ func (css *consensusStateStore) commitVirtualUTXODiff(dbTx model.DBTransaction)
 		if err != nil {
 			return err
 		}
+
+		css.virtualUTXOSetCache.Remove(toRemoveOutpoint)
+
 		dbKey, err := utxoKey(toRemoveOutpoint)
 		if err != nil {
 			return err
@@ -55,6 +58,9 @@ func (css *consensusStateStore) commitVirtualUTXODiff(dbTx model.DBTransaction)
 		if err != nil {
 			return err
 		}
+
+		css.virtualUTXOSetCache.Add(toAddOutpoint, toAddEntry)
+
 		dbKey, err := utxoKey(toAddOutpoint)
 		if err != nil {
 			return err
@@ -79,12 +85,15 @@ func (css *consensusStateStore) commitVirtualUTXOSet(dbTx model.DBTransaction) e
 		return nil
 	}
 
+	css.virtualUTXOSetCache.Clear()
 	iterator := css.virtualUTXOSetStaging.Iterator()
 	for iterator.Next() {
 		outpoint, utxoEntry, err := iterator.Get()
 		if err != nil {
 			return err
 		}
+
+		css.virtualUTXOSetCache.Add(outpoint, utxoEntry)
 		dbKey, err := utxoKey(outpoint)
 		if err != nil {
 			return err
@@ -127,6 +136,10 @@ func (css *consensusStateStore) utxoByOutpointFromStagedVirtualUTXODiff(dbContex
 		}
 	}
 
+	if entry, ok := css.virtualUTXOSetCache.Get(outpoint); ok {
+		return entry, nil
+	}
+
 	key, err := utxoKey(outpoint)
 	if err != nil {
 		return nil, err
@@ -137,7 +150,13 @@ func (css *consensusStateStore) utxoByOutpointFromStagedVirtualUTXODiff(dbContex
 		return nil, err
 	}
 
-	return deserializeUTXOEntry(serializedUTXOEntry)
+	entry, err := deserializeUTXOEntry(serializedUTXOEntry)
+	if err != nil {
+		return nil, err
+	}
+
+	css.virtualUTXOSetCache.Add(outpoint, entry)
+	return entry, nil
 }
 
 func (css *consensusStateStore) utxoByOutpointFromStagedVirtualUTXOSet(outpoint *externalapi.DomainOutpoint) (
diff --git a/infrastructure/db/database/transaction_test.go b/infrastructure/db/database/transaction_test.go
index cfb5247c..eda5b137 100644
--- a/infrastructure/db/database/transaction_test.go
+++ b/infrastructure/db/database/transaction_test.go
@@ -130,15 +130,14 @@ func testTransactionGet(t *testing.T, db database.Database, testName string) {
 			"unexpectedly failed: %s", testName, err)
 	}
 
-	// Make sure that the new value doesn't exist inside the transaction
-	_, err = dbTx.Get(key2)
-	if err == nil {
+	// Make sure that the new value exists inside the transaction
+	newValue2, err := dbTx.Get(key2)
+	if err != nil {
 		t.Fatalf("%s: Get "+
-			"unexpectedly succeeded", testName)
+			"unexpectedly failed: %v", testName, err)
 	}
-	if !database.IsNotFoundError(err) {
-		t.Fatalf("%s: Get "+
-			"returned wrong error: %s", testName, err)
+	if !bytes.Equal(value2, newValue2) {
+		t.Fatalf("Expected %x and %x to be the same", value2, newValue2)
 	}
 
 	// Put a new value into the transaction
@@ -221,15 +220,15 @@ func testTransactionHas(t *testing.T, db database.Database, testName string) {
 			"unexpectedly failed: %s", testName, err)
 	}
 
-	// Make sure that the new value doesn't exist inside the transaction
+	// Make sure that the new value exists inside the transaction
 	exists, err = dbTx.Has(key2)
 	if err != nil {
 		t.Fatalf("%s: Has "+
 			"unexpectedly failed: %s", testName, err)
 	}
-	if exists {
+	if !exists {
 		t.Fatalf("%s: Has "+
-			"unexpectedly returned that the value exists", testName)
+			"unexpectedly returned that the value doesn't exists", testName)
 	}
 }
 
@@ -296,15 +295,15 @@ func testTransactionDelete(t *testing.T, db database.Database, testName string)
 			"unexpectedly returned that the value exists", testName)
 	}
 
-	// Make sure that the second transaction was no affected
+	// Make sure that the second transaction is also affected
 	exists, err = dbTx2.Has(key)
 	if err != nil {
 		t.Fatalf("%s: Has "+
 			"unexpectedly failed: %s", testName, err)
 	}
-	if !exists {
+	if exists {
 		t.Fatalf("%s: Has "+
-			"unexpectedly returned that the value does not exist", testName)
+			"unexpectedly returned that the value exists", testName)
 	}
 }