Merge pull request #2020 from onetechnical/onetechnical/relbeta2.5.3

go-algorand 2.5.3-beta

Author: algojohnlee

buildnumber.dat

@@ -1 +1 @@
-2
+3

data/pools/transactionPool.go

@@ -109,7 +109,7 @@ func MakeTransactionPool(ledger *ledger.Ledger, cfg config.Local, log logging.Lo
 	}
 	pool.cond.L = &pool.mu
 	pool.assemblyCond.L = &pool.assemblyMu
-	pool.recomputeBlockEvaluator(make(map[transactions.Txid]basics.Round))
+	pool.recomputeBlockEvaluator(make(map[transactions.Txid]basics.Round), 0)
 	return &pool
 }
 
@@ -162,7 +162,7 @@ func (pool *TransactionPool) Reset() {
 	pool.numPendingWholeBlocks = 0
 	pool.pendingBlockEvaluator = nil
 	pool.statusCache.reset()
-	pool.recomputeBlockEvaluator(make(map[transactions.Txid]basics.Round))
+	pool.recomputeBlockEvaluator(make(map[transactions.Txid]basics.Round), 0)
 }
 
 // NumExpired returns the number of transactions that expired at the
@@ -468,10 +468,10 @@ func (pool *TransactionPool) OnNewBlock(block bookkeeping.Block, delta ledgercor
 	var knownCommitted uint
 	var unknownCommitted uint
 
-	commitedTxids := delta.Txids
+	committedTxids := delta.Txids
 	if pool.logProcessBlockStats {
 		pool.pendingMu.RLock()
-		for txid := range commitedTxids {
+		for txid := range committedTxids {
 			if _, ok := pool.pendingTxids[txid]; ok {
 				knownCommitted++
 			} else {
@@ -512,7 +512,7 @@ func (pool *TransactionPool) OnNewBlock(block bookkeeping.Block, delta ledgercor
 		// Recompute the pool by starting from the new latest block.
 		// This has the side-effect of discarding transactions that
 		// have been committed (or that are otherwise no longer valid).
-		stats = pool.recomputeBlockEvaluator(commitedTxids)
+		stats = pool.recomputeBlockEvaluator(committedTxids, knownCommitted)
 	}
 
 	stats.KnownCommittedCount = knownCommitted
@@ -625,7 +625,7 @@ func (pool *TransactionPool) addToPendingBlockEvaluator(txgroup []transactions.S
 // recomputeBlockEvaluator constructs a new BlockEvaluator and feeds all
 // in-pool transactions to it (removing any transactions that are rejected
 // by the BlockEvaluator). Expects that the pool.mu mutex would be already taken.
-func (pool *TransactionPool) recomputeBlockEvaluator(committedTxIds map[transactions.Txid]basics.Round) (stats telemetryspec.ProcessBlockMetrics) {
+func (pool *TransactionPool) recomputeBlockEvaluator(committedTxIds map[transactions.Txid]basics.Round, knownCommitted uint) (stats telemetryspec.ProcessBlockMetrics) {
 	pool.pendingBlockEvaluator = nil
 
 	latest := pool.ledger.Latest()
@@ -665,7 +665,11 @@ func (pool *TransactionPool) recomputeBlockEvaluator(committedTxIds map[transact
 
 	next := bookkeeping.MakeBlock(prev)
 	pool.numPendingWholeBlocks = 0
-	pool.pendingBlockEvaluator, err = pool.ledger.StartEvaluator(next.BlockHeader, pendingCount)
+	hint := pendingCount - int(knownCommitted)
+	if hint < 0 || int(knownCommitted) < 0 {
+		hint = 0
+	}
+	pool.pendingBlockEvaluator, err = pool.ledger.StartEvaluator(next.BlockHeader, hint)
 	if err != nil {
 		pool.log.Warnf("TransactionPool.recomputeBlockEvaluator: cannot start evaluator: %v", err)
 		return

go.mod

@@ -8,7 +8,7 @@ require (
 	github.com/algorand/graphtrace v0.0.0-20201117160756-e524ed1a6f64
 	github.com/algorand/msgp v1.1.47
 	github.com/algorand/oapi-codegen v1.3.5-algorand5
-	github.com/algorand/websocket v1.4.1
+	github.com/algorand/websocket v1.4.2
 	github.com/aws/aws-sdk-go v1.16.5
 	github.com/cpuguy83/go-md2man v1.0.8 // indirect
 	github.com/davidlazar/go-crypto v0.0.0-20170701192655-dcfb0a7ac018

go.sum

@@ -35,8 +35,8 @@ github.com/algorand/msgp v1.1.47 h1:xeU6G/Mb1iudJe4L5X38YrOY+VHhvHQDZXxyXYHTzOw=
 github.com/algorand/msgp v1.1.47/go.mod h1:LtOntbYiCHj/Sl/Sqxtf8CZOrDt2a8Dv3tLaS6mcnUE=
 github.com/algorand/oapi-codegen v1.3.5-algorand5 h1:y576Ca2/guQddQrQA7dtL5KcOx5xQgPeIupiuFMGyCI=
 github.com/algorand/oapi-codegen v1.3.5-algorand5/go.mod h1:/k0Ywn0lnt92uBMyE+yiRf/Wo3/chxHHsAfenD09EbY=
-github.com/algorand/websocket v1.4.1 h1:FPoNHI8i2VZWZzhCscY8JTzsAE7Vv73753cMbzb3udk=
-github.com/algorand/websocket v1.4.1/go.mod h1:0nFSn+xppw/GZS9hgWPS3b8/4FcA3Pj7XQxm+wqHGx8=
+github.com/algorand/websocket v1.4.2 h1:zMB7ukz+c7tcef8rVqmKQTv6KQtxXtCFuiAqKaE7n9I=
+github.com/algorand/websocket v1.4.2/go.mod h1:0nFSn+xppw/GZS9hgWPS3b8/4FcA3Pj7XQxm+wqHGx8=
 github.com/armon/circbuf v0.0.0-20150827004946-bbbad097214e/go.mod h1:3U/XgcO3hCbHZ8TKRvWD2dDTCfh9M9ya+I9JpbB7O8o=
 github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da/go.mod h1:Q73ZrmVTwzkszR9V5SSuryQ31EELlFMUz1kKyl939pY=
 github.com/armon/go-radix v0.0.0-20180808171621-7fddfc383310/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8=

ledger/appcow.go

@@ -410,7 +410,7 @@ func MakeDebugBalances(l ledgerForCowBase, round basics.Round, proto protocol.Co
 // Execution happens in a child cow and all modifications are merged into parent if the program passes
 func (cb *roundCowState) StatefulEval(params logic.EvalParams, aidx basics.AppIndex, program []byte) (pass bool, evalDelta basics.EvalDelta, err error) {
 	// Make a child cow to eval our program in
-	calf := cb.child()
+	calf := cb.child(1)
 	params.Ledger, err = newLogicLedger(calf, aidx)
 	if err != nil {
 		return false, basics.EvalDelta{}, err

ledger/appcow_test.go

@@ -305,7 +305,7 @@ func TestCowStorage(t *testing.T) {
 		// Make a child
 		if childDepth < maxChildDepth && rand.Float32() < 0.1 {
 			lastParent = cow
-			cow = cow.child()
+			cow = cow.child(1)
 			childDepth++
 		}
 

ledger/apply/asset.go

@@ -24,15 +24,15 @@ import (
 )
 
 func cloneAssetHoldings(m map[basics.AssetIndex]basics.AssetHolding) map[basics.AssetIndex]basics.AssetHolding {
-	res := make(map[basics.AssetIndex]basics.AssetHolding)
+	res := make(map[basics.AssetIndex]basics.AssetHolding, len(m))
 	for id, val := range m {
 		res[id] = val
 	}
 	return res
 }
 
 func cloneAssetParams(m map[basics.AssetIndex]basics.AssetParams) map[basics.AssetIndex]basics.AssetParams {
-	res := make(map[basics.AssetIndex]basics.AssetParams)
+	res := make(map[basics.AssetIndex]basics.AssetParams, len(m))
 	for id, val := range m {
 		res[id] = val
 	}

ledger/apply/asset_test.go

@@ -17,6 +17,7 @@
 package apply
 
 import (
+	"math/rand"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -97,3 +98,19 @@ func TestAssetTransfer(t *testing.T) {
 		require.Equal(t, dstAmount-toSend, addrs[cls].Assets[1].Amount)
 	}
 }
+
+var benchTotal int = 0
+
+func BenchmarkAssetCloning(b *testing.B) {
+	const numAssets = 800
+	assets := make(map[basics.AssetIndex]basics.AssetHolding, numAssets)
+	for j := 0; j < numAssets; j++ {
+		aidx := basics.AssetIndex(rand.Int63n(100000000))
+		assets[aidx] = basics.AssetHolding{Amount: uint64(aidx)}
+	}
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		clone := cloneAssetHoldings(assets)
+		benchTotal += len(clone) // make sure the compiler does not optimize out cloneAssetHoldings call
+	}
+}

ledger/cow.go

@@ -195,12 +195,12 @@ func (cb *roundCowState) setCompactCertNext(rnd basics.Round) {
 	cb.mods.CompactCertNext = rnd
 }
 
-func (cb *roundCowState) child() *roundCowState {
+func (cb *roundCowState) child(hint int) *roundCowState {
 	return &roundCowState{
 		lookupParent: cb,
 		commitParent: cb,
 		proto:        cb.proto,
-		mods:         ledgercore.MakeStateDelta(cb.mods.Hdr, cb.mods.PrevTimestamp, 1),
+		mods:         ledgercore.MakeStateDelta(cb.mods.Hdr, cb.mods.PrevTimestamp, hint),
 		sdeltas:      make(map[basics.Address]map[storagePtr]*storageDelta),
 	}
 }

ledger/cow_test.go

@@ -105,7 +105,7 @@ func TestCowBalance(t *testing.T) {
 	c0 := makeRoundCowState(&ml, bookkeeping.BlockHeader{}, 0, 0)
 	checkCow(t, c0, accts0)
 
-	c1 := c0.child()
+	c1 := c0.child(0)
 	checkCow(t, c0, accts0)
 	checkCow(t, c1, accts0)
 
@@ -114,7 +114,7 @@ func TestCowBalance(t *testing.T) {
 	checkCow(t, c0, accts0)
 	checkCow(t, c1, accts1)
 
-	c2 := c1.child()
+	c2 := c1.child(0)
 	checkCow(t, c0, accts0)
 	checkCow(t, c1, accts1)
 	checkCow(t, c2, accts1)

ledger/eval.go

@@ -581,7 +581,7 @@ func (eval *BlockEvaluator) TestTransactionGroup(txgroup []transactions.SignedTx
 		return fmt.Errorf("group size %d exceeds maximum %d", len(txgroup), eval.proto.MaxTxGroupSize)
 	}
 
-	cow := eval.state.child()
+	cow := eval.state.child(len(txgroup))
 
 	var group transactions.TxGroup
 	for gi, txn := range txgroup {
@@ -715,7 +715,7 @@ func (eval *BlockEvaluator) transactionGroup(txgroup []transactions.SignedTxnWit
 	var group transactions.TxGroup
 	var groupTxBytes int
 
-	cow := eval.state.child()
+	cow := eval.state.child(len(txgroup))
 
 	// Prepare eval params for any ApplicationCall transactions in the group
 	evalParams := eval.prepareEvalParams(txgroup)
@@ -1009,6 +1009,7 @@ func (eval *BlockEvaluator) endOfBlock() error {
 
 // FinalValidation does the validation that must happen after the block is built and all state updates are computed
 func (eval *BlockEvaluator) finalValidation() error {
+	eval.state.mods.OptimizeAllocatedMemory(eval.proto)
 	if eval.validate {
 		// check commitments
 		txnRoot, err := eval.block.PaysetCommit()

ledger/ledgercore/statedelta.go

@@ -17,11 +17,20 @@
 package ledgercore
 
 import (
+	"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/data/transactions"
 )
 
+const (
+	accountArrayEntrySize                 = uint64(232) // Measured by BenchmarkBalanceRecord
+	accountMapCacheEntrySize              = uint64(64)  // Measured by BenchmarkAcctCache
+	txleasesEntrySize                     = uint64(112) // Measured by BenchmarkTxLeases
+	creatablesEntrySize                   = uint64(100) // Measured by BenchmarkCreatables
+	stateDeltaTargetOptimizationThreshold = uint64(50000000)
+)
+
 // ModifiedCreatable defines the changes to a single single creatable state
 type ModifiedCreatable struct {
 	// Type of the creatable: app or asset
@@ -68,6 +77,9 @@ type StateDelta struct {
 
 	// previous block timestamp
 	PrevTimestamp int64
+
+	// initial hint for allocating data structures for StateDelta
+	initialTransactionsCount int
 }
 
 // AccountDeltas stores ordered accounts and allows fast lookup by address
@@ -78,18 +90,22 @@ type AccountDeltas struct {
 	acctsCache map[basics.Address]int
 }
 
-// MakeStateDelta creates a new instance of StateDelta
+// MakeStateDelta creates a new instance of StateDelta.
+// hint is amount of transactions for evaluation, 2 * hint is for sender and receiver balance records.
+// This does not play well for AssetConfig and ApplicationCall transactions on scale
 func MakeStateDelta(hdr *bookkeeping.BlockHeader, prevTimestamp int64, hint int) StateDelta {
 	return StateDelta{
 		Accts: AccountDeltas{
 			accts:      make([]basics.BalanceRecord, 0, hint*2),
 			acctsCache: make(map[basics.Address]int, hint*2),
 		},
-		Txids:         make(map[transactions.Txid]basics.Round, hint),
-		Txleases:      make(map[Txlease]basics.Round, hint),
-		Creatables:    make(map[basics.CreatableIndex]ModifiedCreatable, hint),
-		Hdr:           hdr,
-		PrevTimestamp: prevTimestamp,
+		Txids:    make(map[transactions.Txid]basics.Round, hint),
+		Txleases: make(map[Txlease]basics.Round, hint),
+		// asset or application creation are considered as rare events so do not pre-allocate space for them
+		Creatables:               make(map[basics.CreatableIndex]ModifiedCreatable),
+		Hdr:                      hdr,
+		PrevTimestamp:            prevTimestamp,
+		initialTransactionsCount: hint,
 	}
 }
 
@@ -149,3 +165,44 @@ func (ad *AccountDeltas) upsert(br basics.BalanceRecord) {
 	}
 	ad.acctsCache[addr] = last
 }
+
+// OptimizeAllocatedMemory by reallocating maps to needed capacity
+// For each data structure, reallocate if it would save us at least 50MB aggregate
+func (sd *StateDelta) OptimizeAllocatedMemory(proto config.ConsensusParams) {
+	// accts takes up 232 bytes per entry, and is saved for 320 rounds
+	if uint64(cap(sd.Accts.accts)-len(sd.Accts.accts))*accountArrayEntrySize*proto.MaxBalLookback > stateDeltaTargetOptimizationThreshold {
+		accts := make([]basics.BalanceRecord, len(sd.Accts.acctsCache))
+		copy(accts, sd.Accts.accts)
+		sd.Accts.accts = accts
+	}
+
+	// acctsCache takes up 64 bytes per entry, and is saved for 320 rounds
+	// realloc if original allocation capacity greater than length of data, and space difference is significant
+	if 2*sd.initialTransactionsCount > len(sd.Accts.acctsCache) &&
+		uint64(2*sd.initialTransactionsCount-len(sd.Accts.acctsCache))*accountMapCacheEntrySize*proto.MaxBalLookback > stateDeltaTargetOptimizationThreshold {
+		acctsCache := make(map[basics.Address]int, len(sd.Accts.acctsCache))
+		for k, v := range sd.Accts.acctsCache {
+			acctsCache[k] = v
+		}
+		sd.Accts.acctsCache = acctsCache
+	}
+
+	// TxLeases takes up 112 bytes per entry, and is saved for 1000 rounds
+	if sd.initialTransactionsCount > len(sd.Txleases) &&
+		uint64(sd.initialTransactionsCount-len(sd.Txleases))*txleasesEntrySize*proto.MaxTxnLife > stateDeltaTargetOptimizationThreshold {
+		txLeases := make(map[Txlease]basics.Round, len(sd.Txleases))
+		for k, v := range sd.Txleases {
+			txLeases[k] = v
+		}
+		sd.Txleases = txLeases
+	}
+
+	// Creatables takes up 100 bytes per entry, and is saved for 320 rounds
+	if uint64(len(sd.Creatables))*creatablesEntrySize*proto.MaxBalLookback > stateDeltaTargetOptimizationThreshold {
+		creatableDeltas := make(map[basics.CreatableIndex]ModifiedCreatable, len(sd.Creatables))
+		for k, v := range sd.Creatables {
+			creatableDeltas[k] = v
+		}
+		sd.Creatables = creatableDeltas
+	}
+}

ledger/ledgercore/statedelta_test.go

@@ -92,3 +92,60 @@ func TestAccountDeltas(t *testing.T) {
 	a.Equal(addr1, address)
 	a.Equal(sample1, data)
 }
+
+func BenchmarkMakeStateDelta(b *testing.B) {
+	hint := 23000
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		MakeStateDelta(nil, 0, hint)
+	}
+}
+
+func BenchmarkBalanceRecord(b *testing.B) {
+	hint := 23000
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		x := make([]basics.BalanceRecord, 0, hint*2)
+		if len(x) > 0 {
+			return
+		}
+	}
+}
+
+func BenchmarkAcctCache(b *testing.B) {
+	hint := 23000
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		x := make(map[basics.Address]int, hint*2)
+		if len(x) > 0 {
+			return
+		}
+	}
+}
+
+func BenchmarkCreatables(b *testing.B) {
+	hint := 23000
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		x := make(map[basics.CreatableIndex]ModifiedCreatable, hint)
+		if len(x) > 0 {
+			return
+		}
+	}
+}
+
+func BenchmarkTxLeases(b *testing.B) {
+	hint := 23000
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		x := make(map[Txlease]basics.Round, hint)
+		if len(x) > 0 {
+			return
+		}
+	}
+}

network/limited_reader_slurper.go

@@ -77,7 +77,7 @@ func (s *LimitedReaderSlurper) Read(reader io.Reader) error {
 					// if we received err == nil and n == 0, we should retry calling the Read function.
 					continue
 				default:
-					// if we recieved a non-io.EOF error, return it.
+					// if we received a non-io.EOF error, return it.
 					return err
 				}
 			}