Merge branch 'feat/supernova-async-exec' into cross-shard-execution-analysis

Author: mradian1

cmd/node/config/config.toml

@@ -574,6 +574,10 @@
     Type = "SizeLRU"
     SizeInBytes = 104857600 # 100MB
 
+[HeaderBodyCacheConfig]
+    Capacity = 1000
+
+
 #PublicKeyPeerId represents the main cache used to map MultiversX block signing public keys to their associated peer id's.
 [PublicKeyPeerId]
     Name = "PublicKeyPeerId"

cmd/termui/presenter/chainInfoGetters.go

@@ -13,6 +13,16 @@ func (psh *PresenterStatusHandler) GetNonce() uint64 {
 	return psh.getFromCacheAsUint64(common.MetricNonce)
 }
 
+// GetLastExecutedNonce will return current last executed nonce of node
+func (psh *PresenterStatusHandler) GetLastExecutedNonce() uint64 {
+	return psh.getFromCacheAsUint64(common.MetricLastExecutedNonce)
+}
+
+// GetProposedNonce will return current proposed nonce of node
+func (psh *PresenterStatusHandler) GetProposedNonce() uint64 {
+	return psh.getFromCacheAsUint64(common.MetricProposedNonce)
+}
+
 // GetIsSyncing will return state of the node
 func (psh *PresenterStatusHandler) GetIsSyncing() uint64 {
 	return psh.getFromCacheAsUint64(common.MetricIsSyncing)
@@ -72,25 +82,14 @@ func (psh *PresenterStatusHandler) CalculateTimeToSynchronize(numMillisecondsRef
 
 	currentSynchronizedRound := psh.GetSynchronizedRound()
 
-	numSynchronizationSpeedHistory := len(psh.synchronizationSpeedHistory)
-
-	sum := uint64(0)
-	for i := 0; i < len(psh.synchronizationSpeedHistory); i++ {
-		sum += psh.synchronizationSpeedHistory[i]
-	}
-
-	speed := float64(0)
-	if numSynchronizationSpeedHistory > 0 {
-		speed = float64(sum*1000) / float64(numSynchronizationSpeedHistory*numMillisecondsRefreshTime)
-	}
-
+	speed := psh.calculateSpeedFromSpeedHistory(numMillisecondsRefreshTime)
 	currentRound := psh.GetCurrentRound()
 	if currentRound < currentSynchronizedRound || speed == 0 {
-		return ""
+		return "Estimating..."
 	}
 
 	remainingRoundsToSynchronize := currentRound - currentSynchronizedRound
-	timeEstimationSeconds := float64(remainingRoundsToSynchronize) / speed
+	timeEstimationSeconds := float64(remainingRoundsToSynchronize) / float64(speed)
 	remainingTime := core.SecondsToHourMinSec(int(timeEstimationSeconds))
 
 	return remainingTime
@@ -100,47 +99,37 @@ func (psh *PresenterStatusHandler) CalculateTimeToSynchronize(numMillisecondsRef
 // how many blocks per second are synchronized
 func (psh *PresenterStatusHandler) CalculateSynchronizationSpeed(numMillisecondsRefreshTime int) uint64 {
 	currentSynchronizedRound := psh.GetSynchronizedRound()
-	if psh.oldRound == 0 {
-		psh.oldRound = currentSynchronizedRound
-		return 0
-	}
-
-	roundsPerSecond := int64(currentSynchronizedRound - psh.oldRound)
-	if roundsPerSecond < 0 {
-		roundsPerSecond = 0
-	}
 
 	if len(psh.synchronizationSpeedHistory) >= maxSpeedHistorySaved {
-		psh.synchronizationSpeedHistory = psh.synchronizationSpeedHistory[1:len(psh.synchronizationSpeedHistory)]
+		psh.synchronizationSpeedHistory = psh.synchronizationSpeedHistory[1:]
 	}
-	psh.synchronizationSpeedHistory = append(psh.synchronizationSpeedHistory, uint64(roundsPerSecond))
+	psh.synchronizationSpeedHistory = append(psh.synchronizationSpeedHistory, currentSynchronizedRound)
 
-	psh.oldRound = currentSynchronizedRound
+	speed := psh.calculateSpeedFromSpeedHistory(numMillisecondsRefreshTime)
 
-	numSyncedBlocks := uint64(0)
-	cumulatedTime := uint64(0)
+	return speed
+}
+
+func (psh *PresenterStatusHandler) calculateSpeedFromSpeedHistory(numMillisecondsRefreshTime int) uint64 {
 	lastIndex := len(psh.synchronizationSpeedHistory) - 1
-	millisecondsInASecond := uint64(1000)
-	for {
-		if lastIndex < 0 {
-			break
-		}
-		if cumulatedTime >= millisecondsInASecond {
-			break
-		}
+	firstIndex := 0
 
-		numSyncedBlocks += psh.synchronizationSpeedHistory[lastIndex]
-		lastIndex--
-		cumulatedTime += uint64(numMillisecondsRefreshTime)
+	if lastIndex <= firstIndex {
+		return 0
 	}
-	if cumulatedTime == 0 || numSyncedBlocks == 0 {
+
+	numSyncedBlocks := psh.synchronizationSpeedHistory[lastIndex] - psh.synchronizationSpeedHistory[firstIndex]
+	cumulatedTimeMs := uint64((lastIndex - firstIndex) * numMillisecondsRefreshTime)
+
+	if cumulatedTimeMs == 0 || numSyncedBlocks == 0 {
 		return 0
 	}
 
-	timeAdjustment := float64(millisecondsInASecond) / float64(cumulatedTime)
-	syncedBlocksAdjustment := timeAdjustment * float64(numSyncedBlocks)
+	numMillisecondsInASecond := 1000.0
+	speed := (float64(numSyncedBlocks) / float64(cumulatedTimeMs)) * numMillisecondsInASecond
+
+	return uint64(speed)
 
-	return uint64(syncedBlocksAdjustment)
 }
 
 // GetNumTxProcessed will return number of processed transactions since node starts

cmd/termui/presenter/chainInfoGetters_test.go

@@ -22,6 +22,28 @@ func TestPresenterStatusHandler_GetNonce(t *testing.T) {
 	assert.Equal(t, nonce, result)
 }
 
+func TestPresenterStatusHandler_GetLastExecutedNonce(t *testing.T) {
+	t.Parallel()
+
+	nonce := uint64(900)
+	presenterStatusHandler := NewPresenterStatusHandler()
+	presenterStatusHandler.SetUInt64Value(common.MetricLastExecutedNonce, nonce)
+	result := presenterStatusHandler.GetLastExecutedNonce()
+
+	assert.Equal(t, nonce, result)
+}
+
+func TestPresenterStatusHandler_GetProposedNonce(t *testing.T) {
+	t.Parallel()
+
+	nonce := uint64(1000)
+	presenterStatusHandler := NewPresenterStatusHandler()
+	presenterStatusHandler.SetUInt64Value(common.MetricProposedNonce, nonce)
+	result := presenterStatusHandler.GetProposedNonce()
+
+	assert.Equal(t, nonce, result)
+}
+
 func TestPresenterStatusHandler_GetIsSyncing(t *testing.T) {
 	t.Parallel()
 
@@ -132,7 +154,11 @@ func TestPresenterStatusHandler_CalculateTimeToSynchronize(t *testing.T) {
 	time.Sleep(time.Second)
 	presenterStatusHandler.SetUInt64Value(common.MetricSynchronizedRound, currentBlockNonce)
 	presenterStatusHandler.SetUInt64Value(common.MetricCurrentRound, probableHighestNonce)
-	presenterStatusHandler.synchronizationSpeedHistory = append(presenterStatusHandler.synchronizationSpeedHistory, synchronizationSpeed)
+	presenterStatusHandler.synchronizationSpeedHistory = []uint64{
+		currentBlockNonce,
+		currentBlockNonce + synchronizationSpeed,
+	}
+
 	synchronizationEstimation := presenterStatusHandler.CalculateTimeToSynchronize(1000)
 
 	// Node needs to synchronize 190 blocks and synchronization speed is 10 blocks/s

cmd/termui/view/interface.go

@@ -29,6 +29,8 @@ type Presenter interface {
 	GetIsSyncing() uint64
 	GetTxPoolLoad() uint64
 	GetNonce() uint64
+	GetProposedNonce() uint64
+	GetLastExecutedNonce() uint64
 	GetProbableHighestNonce() uint64
 	GetSynchronizedRound() uint64
 	GetRoundTime() uint64

cmd/termui/view/termuic/termuiRenders/widgetsRender.go

@@ -303,9 +303,20 @@ func (wr *WidgetsRender) prepareBlockInfo() {
 	numRows := 10
 	rows := make([][]string, numRows)
 
-	currentBlockHeight := wr.presenter.GetNonce()
+	currentBlockHeight := wr.presenter.GetProposedNonce()
 	blockSize := wr.presenter.GetBlockSize()
-	rows[0] = []string{fmt.Sprintf("Current block height: %d, size: %s", currentBlockHeight, core.ConvertBytes(blockSize))}
+	lastExecutedNonce := wr.presenter.GetLastExecutedNonce()
+	lastNotarizedExecutedNonce := wr.presenter.GetNonce()
+
+	if currentBlockHeight == 0 {
+		currentBlockHeight = lastNotarizedExecutedNonce
+	}
+
+	if lastExecutedNonce == 0 {
+		lastExecutedNonce = currentBlockHeight
+	}
+
+	rows[0] = []string{fmt.Sprintf("Current block height: %d, size: %s, last executed nonce: %d, last notarized executed nonce: %d", currentBlockHeight, core.ConvertBytes(blockSize), lastExecutedNonce, lastNotarizedExecutedNonce)}
 
 	numTransactionInBlock := wr.presenter.GetNumTxInBlock()
 	numMiniBlocks := wr.presenter.GetNumMiniBlocks()

cmd/termui/view/termuic/termuiRenders/widgetsRender_test.go

@@ -50,7 +50,9 @@ func TestWidgetsRender_prepareLogLines(t *testing.T) {
 
 func TestPrepareBlockInfo(t *testing.T) {
 	presenterStatusHandler := presenter.NewPresenterStatusHandler()
-	presenterStatusHandler.SetUInt64Value(common.MetricNonce, 42)
+	presenterStatusHandler.SetUInt64Value(common.MetricNonce, 38)
+	presenterStatusHandler.SetUInt64Value(common.MetricLastExecutedNonce, 40)
+	presenterStatusHandler.SetUInt64Value(common.MetricProposedNonce, 42)
 	presenterStatusHandler.SetUInt64Value(common.MetricMiniBlocksSize, 2000)
 	presenterStatusHandler.SetUInt64Value(common.MetricHeaderSize, 48)
 	presenterStatusHandler.SetUInt64Value(common.MetricNumTxInBlock, 5)
@@ -83,7 +85,7 @@ func TestPrepareBlockInfo(t *testing.T) {
 	require.Len(t, wr.blockInfo.Rows, 10)
 
 	// Example: check one row
-	require.Contains(t, wr.blockInfo.Rows[0][0], fmt.Sprintf("Current block height: %d, size: %s", 42, "2.00 KB"))
+	require.Contains(t, wr.blockInfo.Rows[0][0], fmt.Sprintf("Current block height: %d, size: %s, last executed nonce: %d, last notarized executed nonce: %d", 42, "2.00 KB", 40, 38))
 	require.Contains(t, wr.blockInfo.Rows[2][0], "hash")
 	require.Contains(t, wr.blockInfo.Rows[4][0], "Consensus state: ProposedBlock | Consensus round state: Success")
 	require.Contains(t, wr.blockInfo.Rows[5][0], "Received proposed block: 0.002000 sec | Received proof: 0.003000 sec")

common/channels.go

@@ -16,3 +16,16 @@ func CloseKeyValueHolderChan(ch chan core.KeyValueHolder) {
 		close(ch)
 	}
 }
+
+// EmptyUint64Channel will return the number of reads from the channel
+func EmptyUint64Channel(ch chan uint64) int {
+	nrReads := 0
+	for {
+		select {
+		case <-ch:
+			nrReads++
+		default:
+			return nrReads
+		}
+	}
+}

common/channels_test.go

@@ -36,3 +36,40 @@ func didTriggerHappen(ch chan struct{}) bool {
 		return false
 	}
 }
+
+func TestEmptyUint64Channel_EmptyChannel(t *testing.T) {
+	t.Parallel()
+
+	ch := make(chan uint64, 5)
+	nrReads := EmptyUint64Channel(ch)
+
+	require.Equal(t, 0, nrReads)
+}
+
+func TestEmptyUint64Channel_ChannelWithValues(t *testing.T) {
+	t.Parallel()
+
+	ch := make(chan uint64, 5)
+	ch <- 1
+	ch <- 2
+	ch <- 3
+
+	nrReads := EmptyUint64Channel(ch)
+
+	require.Equal(t, 3, nrReads)
+	require.Equal(t, 0, len(ch))
+}
+
+func TestEmptyUint64Channel_FullChannel(t *testing.T) {
+	t.Parallel()
+
+	ch := make(chan uint64, 3)
+	ch <- 10
+	ch <- 20
+	ch <- 30
+
+	nrReads := EmptyUint64Channel(ch)
+
+	require.Equal(t, 3, nrReads)
+	require.Equal(t, 0, len(ch))
+}

common/common.go

@@ -603,51 +603,3 @@ func GetMiniBlockHeadersFromExecResult(header data.HeaderHandler) ([]data.MiniBl
 
 	return mbHeaderHandlers, nil
 }
-
-// GetAccumulatedFeesInEpoch returns the accumulated fees in epoch from the header
-func GetAccumulatedFeesInEpoch(header data.HeaderHandler) *big.Int {
-	if check.IfNil(header) {
-		return big.NewInt(0)
-	}
-
-	if !header.IsHeaderV3() {
-		metaHeader, ok := header.(data.MetaHeaderHandler)
-		if !ok {
-			return big.NewInt(0)
-		}
-
-		return metaHeader.GetAccumulatedFeesInEpoch()
-	}
-
-	metaExecutionResult, ok := header.GetLastExecutionResultHandler().(data.LastMetaExecutionResultHandler)
-	if !ok {
-		log.Warn("GetAccumulatedFeesInEpoch cannot cast last execution result handler to data.LastMetaExecutionResultHandler")
-		return big.NewInt(0)
-	}
-
-	return metaExecutionResult.GetExecutionResultHandler().GetAccumulatedFeesInEpoch()
-}
-
-// GetDeveloperFeesInEpoch returns the developer fees in epoch from the header
-func GetDeveloperFeesInEpoch(header data.HeaderHandler) *big.Int {
-	if check.IfNil(header) {
-		return big.NewInt(0)
-	}
-
-	if !header.IsHeaderV3() {
-		metaHeader, ok := header.(data.MetaHeaderHandler)
-		if !ok {
-			return big.NewInt(0)
-		}
-
-		return metaHeader.GetDevFeesInEpoch()
-	}
-
-	metaExecutionResult, ok := header.GetLastExecutionResultHandler().(data.LastMetaExecutionResultHandler)
-	if !ok {
-		log.Warn("GetDeveloperFeesInEpoch cannot cast last execution result handler to data.LastMetaExecutionResultHandler")
-		return big.NewInt(0)
-	}
-
-	return metaExecutionResult.GetExecutionResultHandler().GetDevFeesInEpoch()
-}