Merge pull request #4533 from ElrondNetwork/intercepted-trie-nodes-re…

…factor

Intercepted trie nodes are no longer saved as unmarshalled entities in cacher

cmd/node/config/config.toml

@@ -903,8 +903,9 @@
 [TrieSync]
     NumConcurrentTrieSyncers  = 200
     MaxHardCapForMissingNodes = 5000
-    #available versions: 1 and 2. 1 is the initial version, 2 is updated, more efficient version
-    TrieSyncerVersion         = 2
+    #available versions: 1, 2 and 3. 1 is the initial version, 2 is updated, more efficient version employing 2 lists
+    #the 3-rd one uses depth-first algorithm which keeps the memory consumption low
+    TrieSyncerVersion         = 3
     CheckNodesOnDisk          = false
 
 [Resolvers]

common/statistics/resourceMonitor.go

@@ -153,7 +153,6 @@ func GetRuntimeStatistics() []interface{} {
 	return []interface{}{
 		"timestamp", time.Now().Unix(),
 		"num go", runtime.NumGoroutine(),
-		"alloc", core.ConvertBytes(memStats.Alloc),
 		"heap alloc", core.ConvertBytes(memStats.HeapAlloc),
 		"heap idle", core.ConvertBytes(memStats.HeapIdle),
 		"heap inuse", core.ConvertBytes(memStats.HeapInuse),

integrationTests/state/stateTrieSync/stateTrieSync_test.go

@@ -47,6 +47,15 @@ func createTestProcessorNodeAndTrieStorage(
 }
 
 func TestNode_RequestInterceptTrieNodesWithMessenger(t *testing.T) {
+	t.Run("test with double lists version", func(t *testing.T) {
+		testNodeRequestInterceptTrieNodesWithMessenger(t, 2)
+	})
+	t.Run("test with depth version", func(t *testing.T) {
+		testNodeRequestInterceptTrieNodesWithMessenger(t, 3)
+	})
+}
+
+func testNodeRequestInterceptTrieNodesWithMessenger(t *testing.T, version int) {
 	if testing.Short() {
 		t.Skip("this is not a short test")
 	}
@@ -120,7 +129,7 @@ func TestNode_RequestInterceptTrieNodesWithMessenger(t *testing.T) {
 		MaxHardCapForMissingNodes: 10000,
 		CheckNodesOnDisk:          false,
 	}
-	trieSyncer, _ := trie.NewDoubleListTrieSyncer(arg)
+	trieSyncer, _ := trie.CreateTrieSyncer(arg, version)
 
 	ctxPrint, cancel := context.WithCancel(context.Background())
 	go printStatistics(ctxPrint, tss)
@@ -179,6 +188,15 @@ func printStatistics(ctx context.Context, stats common.SizeSyncStatisticsHandler
 }
 
 func TestNode_RequestInterceptTrieNodesWithMessengerNotSyncingShouldErr(t *testing.T) {
+	t.Run("test with double lists version", func(t *testing.T) {
+		testNodeRequestInterceptTrieNodesWithMessengerNotSyncingShouldErr(t, 2)
+	})
+	t.Run("test with depth version", func(t *testing.T) {
+		testNodeRequestInterceptTrieNodesWithMessengerNotSyncingShouldErr(t, 3)
+	})
+}
+
+func testNodeRequestInterceptTrieNodesWithMessengerNotSyncingShouldErr(t *testing.T, version int) {
 	if testing.Short() {
 		t.Skip("this is not a short test")
 	}
@@ -251,7 +269,7 @@ func TestNode_RequestInterceptTrieNodesWithMessengerNotSyncingShouldErr(t *testi
 		MaxHardCapForMissingNodes: 10000,
 		CheckNodesOnDisk:          false,
 	}
-	trieSyncer, _ := trie.NewDoubleListTrieSyncer(arg)
+	trieSyncer, _ := trie.CreateTrieSyncer(arg, version)
 
 	ctxPrint, cancel := context.WithCancel(context.Background())
 	go printStatistics(ctxPrint, tss)
@@ -280,18 +298,28 @@ func TestMultipleDataTriesSyncSmallValues(t *testing.T) {
 		t.Skip("this is not a short test")
 	}
 
-	testMultipleDataTriesSync(t, 1000, 50, 32)
+	t.Run("test with double lists version", func(t *testing.T) {
+		testMultipleDataTriesSync(t, 1000, 50, 32, 2)
+	})
+	t.Run("test with depth version", func(t *testing.T) {
+		testMultipleDataTriesSync(t, 1000, 50, 32, 3)
+	})
 }
 
 func TestMultipleDataTriesSyncLargeValues(t *testing.T) {
 	if testing.Short() {
 		t.Skip("this is not a short test")
 	}
 
-	testMultipleDataTriesSync(t, 3, 3, 1<<21)
+	t.Run("test with double lists version", func(t *testing.T) {
+		testMultipleDataTriesSync(t, 3, 3, 1<<21, 2)
+	})
+	t.Run("test with depth version", func(t *testing.T) {
+		testMultipleDataTriesSync(t, 3, 3, 1<<21, 3)
+	})
 }
 
-func testMultipleDataTriesSync(t *testing.T, numAccounts int, numDataTrieLeaves int, valSize int) {
+func testMultipleDataTriesSync(t *testing.T, numAccounts int, numDataTrieLeaves int, valSize int, version int) {
 	if testing.Short() {
 		t.Skip("this is not a short test")
 	}
@@ -354,7 +382,7 @@ func testMultipleDataTriesSync(t *testing.T, numAccounts int, numDataTrieLeaves
 			Cacher:                    nRequester.DataPool.TrieNodes(),
 			MaxTrieLevelInMemory:      200,
 			MaxHardCapForMissingNodes: 5000,
-			TrieSyncerVersion:         2,
+			TrieSyncerVersion:         version,
 			CheckNodesOnDisk:          false,
 		},
 		ShardId:                shardID,

integrationTests/testProcessorNode.go

@@ -2847,7 +2847,7 @@ func (tpn *TestProcessorNode) initRoundHandler() {
 }
 
 func (tpn *TestProcessorNode) initRequestedItemsHandler() {
-	tpn.RequestedItemsHandler = timecache.NewTimeCache(roundDuration)
+	tpn.RequestedItemsHandler = timecache.NewTimeCache(time.Second)
 }
 
 func (tpn *TestProcessorNode) initBlockTracker() {

node/nodeRunner.go

@@ -1038,6 +1038,7 @@ func (nr *nodeRunner) CreateManagedProcessComponents(
 	}
 
 	log.Trace("creating time cache for requested items components")
+	// TODO consider lowering this (perhaps to 1 second) and use a common const
 	requestedItemsHandler := timecache.NewTimeCache(
 		time.Duration(uint64(time.Millisecond) * coreComponents.GenesisNodesSetup().GetRoundDuration()))
 

process/block/preprocess/gasComputation.go

@@ -11,6 +11,8 @@ import (
 	"github.com/ElrondNetwork/elrond-go/process"
 )
 
+const initialAllocation = 1000
+
 var _ process.GasHandler = (*gasComputation)(nil)
 
 type gasComputation struct {
@@ -91,18 +93,19 @@ func (gc *gasComputation) Init() {
 }
 
 // Reset method resets tx hashes with gas provided, refunded and penalized since last reset
+// TODO remove this call from basePreProcess.handleProcessTransactionInit
 func (gc *gasComputation) Reset(key []byte) {
 	gc.mutGasProvided.Lock()
-	gc.txHashesWithGasProvidedSinceLastReset[string(key)] = make([][]byte, 0)
-	gc.txHashesWithGasProvidedAsScheduledSinceLastReset[string(key)] = make([][]byte, 0)
+	gc.txHashesWithGasProvidedSinceLastReset[string(key)] = make([][]byte, 0, initialAllocation)
+	gc.txHashesWithGasProvidedAsScheduledSinceLastReset[string(key)] = make([][]byte, 0, initialAllocation)
 	gc.mutGasProvided.Unlock()
 
 	gc.mutGasRefunded.Lock()
-	gc.txHashesWithGasRefundedSinceLastReset[string(key)] = make([][]byte, 0)
+	gc.txHashesWithGasRefundedSinceLastReset[string(key)] = make([][]byte, 0, initialAllocation)
 	gc.mutGasRefunded.Unlock()
 
 	gc.mutGasPenalized.Lock()
-	gc.txHashesWithGasPenalizedSinceLastReset[string(key)] = make([][]byte, 0)
+	gc.txHashesWithGasPenalizedSinceLastReset[string(key)] = make([][]byte, 0, initialAllocation)
 	gc.mutGasPenalized.Unlock()
 }
 

process/interceptors/factory/interceptedTrieNodeDataFactory.go

@@ -3,16 +3,14 @@ package factory
 import (
 	"github.com/ElrondNetwork/elrond-go-core/core/check"
 	"github.com/ElrondNetwork/elrond-go-core/hashing"
-	"github.com/ElrondNetwork/elrond-go-core/marshal"
 	"github.com/ElrondNetwork/elrond-go/process"
 	"github.com/ElrondNetwork/elrond-go/trie"
 )
 
 var _ process.InterceptedDataFactory = (*interceptedTrieNodeDataFactory)(nil)
 
 type interceptedTrieNodeDataFactory struct {
-	marshalizer marshal.Marshalizer
-	hasher      hashing.Hasher
+	hasher hashing.Hasher
 }
 
 // NewInterceptedTrieNodeDataFactory creates an instance of interceptedTrieNodeDataFactory
@@ -26,22 +24,18 @@ func NewInterceptedTrieNodeDataFactory(
 	if check.IfNil(argument.CoreComponents) {
 		return nil, process.ErrNilCoreComponentsHolder
 	}
-	if check.IfNil(argument.CoreComponents.InternalMarshalizer()) {
-		return nil, process.ErrNilMarshalizer
-	}
 	if check.IfNil(argument.CoreComponents.Hasher()) {
 		return nil, process.ErrNilHasher
 	}
 
 	return &interceptedTrieNodeDataFactory{
-		marshalizer: argument.CoreComponents.InternalMarshalizer(),
-		hasher:      argument.CoreComponents.Hasher(),
+		hasher: argument.CoreComponents.Hasher(),
 	}, nil
 }
 
 // Create creates instances of InterceptedData by unmarshalling provided buffer
 func (sidf *interceptedTrieNodeDataFactory) Create(buff []byte) (process.InterceptedData, error) {
-	return trie.NewInterceptedTrieNode(buff, sidf.marshalizer, sidf.hasher)
+	return trie.NewInterceptedTrieNode(buff, sidf.hasher)
 }
 
 // IsInterfaceNil returns true if there is no value under the interface

process/interceptors/factory/interceptedTrieNodeDataFactory_test.go

@@ -16,18 +16,6 @@ func TestNewInterceptedTrieNodeDataFactory_NilArgumentsShouldErr(t *testing.T) {
 	assert.Equal(t, process.ErrNilArgumentStruct, err)
 }
 
-func TestNewInterceptedTrieNodeDataFactory_NilMarshalizerShouldErr(t *testing.T) {
-	t.Parallel()
-
-	coreComponents, cryptoComponents := createMockComponentHolders()
-	coreComponents.IntMarsh = nil
-	arg := createMockArgument(coreComponents, cryptoComponents)
-
-	itn, err := NewInterceptedTrieNodeDataFactory(arg)
-	assert.Nil(t, itn)
-	assert.Equal(t, process.ErrNilMarshalizer, err)
-}
-
 func TestNewInterceptedTrieNodeDataFactory_NilHasherShouldErr(t *testing.T) {
 	t.Parallel()
 

state/syncer/baseAccountsSyncer.go

@@ -12,6 +12,7 @@ import (
 	"github.com/ElrondNetwork/elrond-go-core/hashing"
 	"github.com/ElrondNetwork/elrond-go-core/marshal"
 	"github.com/ElrondNetwork/elrond-go/common"
+	"github.com/ElrondNetwork/elrond-go/common/statistics"
 	"github.com/ElrondNetwork/elrond-go/state"
 	"github.com/ElrondNetwork/elrond-go/storage"
 	"github.com/ElrondNetwork/elrond-go/trie"
@@ -147,6 +148,7 @@ func (b *baseAccountsSyncer) printStatistics(ssh common.SizeSyncStatisticsHandle
 				"peak processing speed", peakSpeed,
 				"average processing speed", averageSpeed,
 			)
+			log.Debug("trie sync node statistics", statistics.GetRuntimeStatistics()...)
 			return
 		case <-time.After(timeBetweenStatisticsPrints):
 			bytesReceivedDelta := ssh.NumBytesReceived() - lastDataReceived
@@ -174,6 +176,7 @@ func (b *baseAccountsSyncer) printStatistics(ssh common.SizeSyncStatisticsHandle
 				"iterations", ssh.NumIterations(),
 				"CPU time", ssh.ProcessingTime(),
 				"processing speed", speed)
+			log.Debug("trie sync node statistics", statistics.GetRuntimeStatistics()...)
 		}
 	}
 }

trie/branchNode_test.go

@@ -1069,7 +1069,7 @@ func TestBranchNode_loadChildren(t *testing.T) {
 	nodes, _ := getEncodedTrieNodesAndHashes(tr)
 	nodesCacher, _ := lrucache.NewCache(100)
 	for i := range nodes {
-		n, _ := NewInterceptedTrieNode(nodes[i], marsh, hasher)
+		n, _ := NewInterceptedTrieNode(nodes[i], hasher)
 		nodesCacher.Put(n.hash, n, len(n.GetSerialized()))
 	}