Message validation optimization

integrationTests/factory/bootstrapComponents/bootstrapComponents_test.go

@@ -10,8 +10,6 @@ import (
 
 	"github.com/multiversx/mx-chain-go/integrationTests/factory"
 	"github.com/multiversx/mx-chain-go/node"
-	//"github.com/multiversx/mx-chain-go/process"
-	//factory2 "github.com/multiversx/mx-chain-go/process/interceptors/factory"
 	"github.com/multiversx/mx-chain-go/process/mock"
 	"github.com/multiversx/mx-chain-go/testscommon/goroutines"
 )

process/factory/interceptorscontainer/baseInterceptorsContainerFactory.go

@@ -958,17 +958,3 @@ func (bicf *baseInterceptorsContainerFactory) addInterceptorsToContainers(keys [
 
 	return bicf.fullArchiveContainer.AddMultiple(keys, interceptors)
 }
-
-//func (bicf *baseInterceptorsContainerFactory) createCacheForInterceptor(topic string) (process.InterceptedDataVerifier, error) {
-//	internalCache, err := cache.NewTimeCacher(cache.ArgTimeCacher{
-//		DefaultSpan: cacheDefaultSpan,
-//		CacheExpiry: cacheDefaultExpiry,
-//	})
-//	if err != nil {
-//		return nil, err
-//	}
-//
-//	bicf.processedMessagesCacheMap[topic] = internalCache
-//	verifier := interceptors.NewInterceptedDataVerifier(internalCache)
-//	return verifier, nil
-//}

process/interceptors/factory/interceptedDataVerifierFactory.go

@@ -46,3 +46,8 @@ func (idvf *InterceptedDataVerifierFactory) Create(topic string) (process.Interc
 	verifier := interceptors.NewInterceptedDataVerifier(internalCache)
 	return verifier, nil
 }
+
+// IsInterfaceNil returns true if there is no value under the interface
+func (idvf *InterceptedDataVerifierFactory) IsInterfaceNil() bool {
+	return idvf == nil
+}

process/interceptors/interceptedDataVerifier.go

@@ -9,11 +9,13 @@ import (
 	"github.com/multiversx/mx-chain-go/storage"
 )
 
-type interceptedDataStatus int
+type interceptedDataStatus int8
 
 const (
 	ValidInterceptedData interceptedDataStatus = iota
 	InvalidInterceptedData
+
+	interceptedDataStatusBytesSize = 8
 )
 
 var (
@@ -39,38 +41,38 @@ func NewInterceptedDataVerifier(cache storage.Cacher) *interceptedDataVerifier {
 // It will retrieve the status in the cache if it exists, otherwise it will validate it and store the status of the
 // validation in the cache. Note that the entries are stored for a set period of time
 func (idv *interceptedDataVerifier) Verify(interceptedData process.InterceptedData) error {
+	hash := string(interceptedData.Hash())
+
 	if len(interceptedData.Hash()) == 0 {
 		return interceptedData.CheckValidity()
 	}
 
-	if val, ok := idv.cache.Get(interceptedData.Hash()); ok {
+	idv.km.RLock(hash)
+	val, ok := idv.cache.Get(interceptedData.Hash())
+	idv.km.RUnlock(hash)
+
+	if ok {
 		if val == ValidInterceptedData {
 			return nil
 		}
 
 		return ErrInvalidInterceptedData
 	}
 
-	err := idv.checkValidity(interceptedData)
+	err := interceptedData.CheckValidity()
 	if err != nil {
-		idv.cache.Put(interceptedData.Hash(), InvalidInterceptedData, 8)
+		idv.km.Lock(hash)
+		idv.cache.Put(interceptedData.Hash(), InvalidInterceptedData, interceptedDataStatusBytesSize)
+		idv.km.Unlock(hash)
+
 		return ErrInvalidInterceptedData
 	}
 
-	idv.cache.Put(interceptedData.Hash(), ValidInterceptedData, 100)
+	idv.cache.Put(interceptedData.Hash(), ValidInterceptedData, interceptedDataStatusBytesSize)
 	return nil
 }
 
 // IsInterfaceNil returns true if there is no value under the interface
 func (idv *interceptedDataVerifier) IsInterfaceNil() bool {
 	return idv == nil
 }
-
-func (idv *interceptedDataVerifier) checkValidity(interceptedData process.InterceptedData) error {
-	hash := string(interceptedData.Hash())
-
-	idv.km.Lock(hash)
-	defer idv.km.Unlock(hash)
-
-	return interceptedData.CheckValidity()
-}

process/interceptors/interceptedDataVerifier_test.go

@@ -28,8 +28,11 @@ func defaultInterceptedDataVerifier(span time.Duration) process.InterceptedDataV
 func TestInterceptedDataVerifier_CheckValidityShouldWork(t *testing.T) {
 	t.Parallel()
 
+	checkValidityCounter := atomic.Counter{}
+
 	interceptedData := &testscommon.InterceptedDataStub{
 		CheckValidityCalled: func() error {
+			checkValidityCounter.Add(1)
 			return nil
 		},
 		IsForCurrentShardCalled: func() bool {
@@ -60,13 +63,17 @@ func TestInterceptedDataVerifier_CheckValidityShouldWork(t *testing.T) {
 	wg.Wait()
 
 	require.Equal(t, int64(0), errCount.Get())
+	require.Equal(t, int64(1), checkValidityCounter.Get())
 }
 
 func TestInterceptedDataVerifier_CheckValidityShouldNotWork(t *testing.T) {
 	t.Parallel()
 
+	checkValidityCounter := atomic.Counter{}
+
 	interceptedData := &testscommon.InterceptedDataStub{
 		CheckValidityCalled: func() error {
+			checkValidityCounter.Add(1)
 			return nil
 		},
 		IsForCurrentShardCalled: func() bool {
@@ -93,13 +100,16 @@ func TestInterceptedDataVerifier_CheckValidityShouldNotWork(t *testing.T) {
 
 	err := verifier.Verify(interceptedDataWithErr)
 	require.Equal(t, ErrInvalidInterceptedData, err)
+	require.Equal(t, int64(0), checkValidityCounter.Get())
 
 	err = verifier.Verify(interceptedData)
 	// It is still invalid because it has the same hash.
 	require.Equal(t, ErrInvalidInterceptedData, err)
+	require.Equal(t, int64(0), checkValidityCounter.Get())
 
 	<-time.After(defaultSpan + 100*time.Millisecond)
 
 	err = verifier.Verify(interceptedData)
-	require.Nil(t, err)
+	require.NoError(t, err)
+	require.Equal(t, int64(1), checkValidityCounter.Get())
 }

process/interface.go

@@ -1409,4 +1409,5 @@ type InterceptedDataVerifier interface {
 
 type InterceptedDataVerifierFactory interface {
 	Create(topic string) (InterceptedDataVerifier, error)
+	IsInterfaceNil() bool
 }

process/mock/interceptedDataVerifierFactoryStub.go

@@ -17,3 +17,8 @@ func (idvfs *InterceptedDataVerifierFactoryStub) Create(topic string) (process.I
 
 	return nil, nil
 }
+
+// IsInterfaceNil -
+func (idvfs *InterceptedDataVerifierFactoryStub) IsInterfaceNil() bool {
+	return idvfs == nil
+}