Merge pull request #1864 from ElrondNetwork/bugFix/sync-multiple-tries

Bug fix/sync multiple tries
multiversx · Jun 5, 2020 · 5f2bb92 · 5f2bb92
2 parents 0422e85 + 21eeac9
commit 5f2bb92
Show file tree

Hide file tree

Showing 44 changed files with 472 additions and 249 deletions.
diff --git a/api/mock/cacherStub.go b/api/mock/cacherStub.go
@@ -2,18 +2,24 @@ package mock
 
 // CacherStub -
 type CacherStub struct {
-	ClearCalled           func()
-	PutCalled             func(key []byte, value interface{}) (evicted bool)
-	GetCalled             func(key []byte) (value interface{}, ok bool)
-	HasCalled             func(key []byte) bool
-	PeekCalled            func(key []byte) (value interface{}, ok bool)
-	HasOrAddCalled        func(key []byte, value interface{}) (ok, evicted bool)
-	RemoveCalled          func(key []byte)
-	RemoveOldestCalled    func()
-	KeysCalled            func() [][]byte
-	LenCalled             func() int
-	MaxSizeCalled         func() int
-	RegisterHandlerCalled func(func(key []byte, value interface{}))
+	ClearCalled             func()
+	PutCalled               func(key []byte, value interface{}) (evicted bool)
+	GetCalled               func(key []byte) (value interface{}, ok bool)
+	HasCalled               func(key []byte) bool
+	PeekCalled              func(key []byte) (value interface{}, ok bool)
+	HasOrAddCalled          func(key []byte, value interface{}) (ok, evicted bool)
+	RemoveCalled            func(key []byte)
+	RemoveOldestCalled      func()
+	KeysCalled              func() [][]byte
+	LenCalled               func() int
+	MaxSizeCalled           func() int
+	RegisterHandlerCalled   func(func(key []byte, value interface{}))
+	UnRegisterHandlerCalled func(func(key []byte, value interface{}))
+}
+
+// UnRegisterHandler -
+func (cs *CacherStub) UnRegisterHandler(handler func(key []byte, value interface{})) {
+	cs.UnRegisterHandlerCalled(handler)
 }
 
 // Clear -

diff --git a/cmd/node/config/config.toml b/cmd/node/config/config.toml
@@ -242,9 +242,9 @@
     Shards = 16
 
 [TrieNodesDataPool]
-    Capacity = 50000
+    Capacity = 900000
     Type = "SizeLRU"
-    SizeInBytes = 104857600 #100MB
+    SizeInBytes = 314572800 #300MB
 
 [WhiteListPool]
     Capacity = 100000

diff --git a/consensus/mock/cacherMock.go b/consensus/mock/cacherMock.go
@@ -110,8 +110,11 @@ func (cm *CacherMock) MaxSize() int {
 }
 
 // RegisterHandler -
-func (cm *CacherMock) RegisterHandler(func(key []byte, value interface{})) {
-	panic("implement me")
+func (cm *CacherMock) RegisterHandler(func(key []byte, value interface{}), string) {
+}
+
+// UnRegisterHandler -
+func (cm *CacherMock) UnRegisterHandler(_ string) {
 }
 
 // IsInterfaceNil returns true if there is no value under the interface

diff --git a/core/common.go b/core/common.go
@@ -1,5 +1,9 @@
 package core
 
+import (
+	"crypto/rand"
+)
+
 // EmptyChannel empties the given channel
 func EmptyChannel(ch chan bool) int {
 	readsCnt := 0
@@ -12,3 +16,10 @@ func EmptyChannel(ch chan bool) int {
 		}
 	}
 }
+
+// UniqueIdentifier returns a unique string identifier of 32 bytes
+func UniqueIdentifier() string {
+	buff := make([]byte, 32)
+	_, _ = rand.Read(buff)
+	return string(buff)
+}
diff --git a/data/errors.go b/data/errors.go
@@ -30,3 +30,9 @@ var ErrInvalidCacheSize = errors.New("cache size is invalid")
 
 // ErrInvalidValue signals that an invalid value has been provided such as NaN to an integer field
 var ErrInvalidValue = errors.New("invalid value")
+
+// ErrNilThrottler signals that nil throttler has been provided
+var ErrNilThrottler = errors.New("nil throttler")
+
+// ErrTimeIsOut signals that time is out
+var ErrTimeIsOut = errors.New("time is out")
diff --git a/data/interface.go b/data/interface.go
@@ -232,3 +232,11 @@ type ShardValidatorInfoHandler interface {
 	String() string
 	IsInterfaceNil() bool
 }
+
+// GoRoutineThrottler can monitor the number of the currently running go routines
+type GoRoutineThrottler interface {
+	CanProcess() bool
+	StartProcessing()
+	EndProcessing()
+	IsInterfaceNil() bool
+}
diff --git a/data/mock/cacherMock.go b/data/mock/cacherMock.go
@@ -103,7 +103,12 @@ func (cm *CacherMock) MaxSize() int {
 }
 
 // RegisterHandler -
-func (cm *CacherMock) RegisterHandler(func(key []byte, value interface{})) {
+func (cm *CacherMock) RegisterHandler(func(key []byte, value interface{}), string) {
+	return
+}
+
+// UnRegisterHandler -
+func (cm *CacherMock) UnRegisterHandler(string) {
 	return
 }
 

diff --git a/data/mock/uint64CacherStub.go b/data/mock/uint64CacherStub.go
@@ -2,17 +2,23 @@ package mock
 
 // Uint64CacherStub -
 type Uint64CacherStub struct {
-	ClearCalled           func()
-	PutCalled             func(uint64, interface{}) bool
-	GetCalled             func(uint64) (interface{}, bool)
-	HasCalled             func(uint64) bool
-	PeekCalled            func(uint64) (interface{}, bool)
-	HasOrAddCalled        func(uint64, interface{}) (bool, bool)
-	RemoveCalled          func(uint64)
-	RemoveOldestCalled    func()
-	KeysCalled            func() []uint64
-	LenCalled             func() int
-	RegisterHandlerCalled func(handler func(nonce uint64))
+	ClearCalled             func()
+	PutCalled               func(uint64, interface{}) bool
+	GetCalled               func(uint64) (interface{}, bool)
+	HasCalled               func(uint64) bool
+	PeekCalled              func(uint64) (interface{}, bool)
+	HasOrAddCalled          func(uint64, interface{}) (bool, bool)
+	RemoveCalled            func(uint64)
+	RemoveOldestCalled      func()
+	KeysCalled              func() []uint64
+	LenCalled               func() int
+	RegisterHandlerCalled   func(handler func(nonce uint64))
+	UnRegisterHandlerCalled func(func(key []byte, value interface{}))
+}
+
+// UnRegisterHandler -
+func (ucs *Uint64CacherStub) UnRegisterHandler(handler func(key []byte, value interface{})) {
+	ucs.UnRegisterHandlerCalled(handler)
 }
 
 // Clear -

diff --git a/data/syncer/userAccountsSyncer.go b/data/syncer/userAccountsSyncer.go
@@ -2,8 +2,11 @@ package syncer
 
 import (
 	"context"
+	"sync"
+	"time"
 
 	logger "github.com/ElrondNetwork/elrond-go-logger"
+	"github.com/ElrondNetwork/elrond-go/core/check"
 	"github.com/ElrondNetwork/elrond-go/data"
 	"github.com/ElrondNetwork/elrond-go/data/state"
 	"github.com/ElrondNetwork/elrond-go/data/trie"
@@ -15,14 +18,19 @@ var _ epochStart.AccountsDBSyncer = (*userAccountsSyncer)(nil)
 
 var log = logger.GetOrCreate("syncer")
 
+const timeBetweenRetries = 100 * time.Millisecond
+
 type userAccountsSyncer struct {
 	*baseAccountsSyncer
+	throttler   data.GoRoutineThrottler
+	syncerMutex sync.Mutex
 }
 
 // ArgsNewUserAccountsSyncer defines the arguments needed for the new account syncer
 type ArgsNewUserAccountsSyncer struct {
 	ArgsNewBaseAccountsSyncer
-	ShardId uint32
+	ShardId   uint32
+	Throttler data.GoRoutineThrottler
 }
 
 // NewUserAccountsSyncer creates a user account syncer
@@ -32,6 +40,10 @@ func NewUserAccountsSyncer(args ArgsNewUserAccountsSyncer) (*userAccountsSyncer,
 		return nil, err
 	}
 
+	if check.IfNil(args.Throttler) {
+		return nil, data.ErrNilThrottler
+	}
+
 	b := &baseAccountsSyncer{
 		hasher:               args.Hasher,
 		marshalizer:          args.Marshalizer,
@@ -48,6 +60,7 @@ func NewUserAccountsSyncer(args ArgsNewUserAccountsSyncer) (*userAccountsSyncer,
 
 	u := &userAccountsSyncer{
 		baseAccountsSyncer: b,
+		throttler:          args.Throttler,
 	}
 
 	return u, nil
@@ -81,25 +94,71 @@ func (u *userAccountsSyncer) SyncAccounts(rootHash []byte) error {
 }
 
 func (u *userAccountsSyncer) syncAccountDataTries(rootHashes [][]byte, ctx context.Context) error {
+	var errFound error
+	errMutex := sync.Mutex{}
+
+	wg := sync.WaitGroup{}
+	wg.Add(len(rootHashes))
+
 	for _, rootHash := range rootHashes {
-		dataTrie, err := trie.NewTrie(u.trieStorageManager, u.marshalizer, u.hasher, u.maxTrieLevelInMemory)
-		if err != nil {
-			return err
+		for {
+			if u.throttler.CanProcess() {
+				break
+			}
+
+			select {
+			case <-time.After(timeBetweenRetries):
+				continue
+			case <-ctx.Done():
+				return data.ErrTimeIsOut
+			}
 		}
 
-		u.dataTries[string(rootHash)] = dataTrie
-		trieSyncer, err := trie.NewTrieSyncer(u.requestHandler, u.cacher, dataTrie, u.shardId, factory.AccountTrieNodesTopic)
-		if err != nil {
-			return err
-		}
-		u.trieSyncers[string(rootHash)] = trieSyncer
+		go func(trieRootHash []byte) {
+			newErr := u.syncDataTrie(trieRootHash, ctx)
+			if newErr != nil {
+				errMutex.Lock()
+				errFound = newErr
+				errMutex.Unlock()
+			}
+			wg.Done()
+		}(rootHash)
+	}
 
-		err = trieSyncer.StartSyncing(rootHash, ctx)
-		if err != nil {
-			return err
-		}
+	wg.Wait()
+
+	errMutex.Lock()
+	defer errMutex.Unlock()
+
+	return errFound
+}
+
+func (u *userAccountsSyncer) syncDataTrie(rootHash []byte, ctx context.Context) error {
+	u.throttler.StartProcessing()
+
+	u.syncerMutex.Lock()
+	dataTrie, err := trie.NewTrie(u.trieStorageManager, u.marshalizer, u.hasher, u.maxTrieLevelInMemory)
+	if err != nil {
+		u.syncerMutex.Unlock()
+		return err
 	}
 
+	u.dataTries[string(rootHash)] = dataTrie
+	trieSyncer, err := trie.NewTrieSyncer(u.requestHandler, u.cacher, dataTrie, u.shardId, factory.AccountTrieNodesTopic)
+	if err != nil {
+		u.syncerMutex.Unlock()
+		return err
+	}
+	u.trieSyncers[string(rootHash)] = trieSyncer
+	u.syncerMutex.Unlock()
+
+	err = trieSyncer.StartSyncing(rootHash, ctx)
+	if err != nil {
+		return err
+	}
+
+	u.throttler.EndProcessing()
+
 	return nil
 }
 

diff --git a/data/trie/sync.go b/data/trie/sync.go
@@ -6,6 +6,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/ElrondNetwork/elrond-go/core"
 	"github.com/ElrondNetwork/elrond-go/core/check"
 	"github.com/ElrondNetwork/elrond-go/data"
 	"github.com/ElrondNetwork/elrond-go/storage"
@@ -29,6 +30,7 @@ type trieSyncer struct {
 	requestHandler          RequestHandler
 	interceptedNodes        storage.Cacher
 	mutOperation            sync.RWMutex
+	handlerID               string
 }
 
 const maxNewMissingAddedPerTurn = 10
@@ -67,8 +69,8 @@ func NewTrieSyncer(
 		topic:                   topic,
 		shardId:                 shardId,
 		waitTimeBetweenRequests: time.Second,
+		handlerID:               core.UniqueIdentifier(),
 	}
-	ts.interceptedNodes.RegisterHandler(ts.trieNodeIntercepted)
 
 	return ts, nil
 }

diff --git a/dataRetriever/mock/cacherStub.go b/dataRetriever/mock/cacherStub.go
@@ -2,18 +2,24 @@ package mock
 
 // CacherStub -
 type CacherStub struct {
-	ClearCalled           func()
-	PutCalled             func(key []byte, value interface{}, sizeInBytes int) (evicted bool)
-	GetCalled             func(key []byte) (value interface{}, ok bool)
-	HasCalled             func(key []byte) bool
-	PeekCalled            func(key []byte) (value interface{}, ok bool)
-	HasOrAddCalled        func(key []byte, value interface{}, sizeInBytes int) (ok, evicted bool)
-	RemoveCalled          func(key []byte)
-	RemoveOldestCalled    func()
-	KeysCalled            func() [][]byte
-	LenCalled             func() int
-	MaxSizeCalled         func() int
-	RegisterHandlerCalled func(func(key []byte, value interface{}))
+	ClearCalled             func()
+	PutCalled               func(key []byte, value interface{}, sizeInBytes int) (evicted bool)
+	GetCalled               func(key []byte) (value interface{}, ok bool)
+	HasCalled               func(key []byte) bool
+	PeekCalled              func(key []byte) (value interface{}, ok bool)
+	HasOrAddCalled          func(key []byte, value interface{}, sizeInBytes int) (ok, evicted bool)
+	RemoveCalled            func(key []byte)
+	RemoveOldestCalled      func()
+	KeysCalled              func() [][]byte
+	LenCalled               func() int
+	MaxSizeCalled           func() int
+	RegisterHandlerCalled   func(func(key []byte, value interface{}))
+	UnRegisterHandlerCalled func(id string)
+}
+
+// UnRegisterHandler -
+func (cs *CacherStub) UnRegisterHandler(id string) {
+	cs.UnRegisterHandlerCalled(id)
 }
 
 // Clear -
@@ -67,7 +73,7 @@ func (cs *CacherStub) MaxSize() int {
 }
 
 // RegisterHandler -
-func (cs *CacherStub) RegisterHandler(handler func(key []byte, value interface{})) {
+func (cs *CacherStub) RegisterHandler(handler func(key []byte, value interface{}), _ string) {
 	cs.RegisterHandlerCalled(handler)
 }