Fixes for parallel TX and metrics

app/abci.go

@@ -2,8 +2,10 @@ package app
 
 import (
 	"context"
+	"time"
 
 	sdk "github.com/cosmos/cosmos-sdk/types"
+	"github.com/sei-protocol/sei-chain/utils/metrics"
 	abci "github.com/tendermint/tendermint/abci/types"
 	"go.opentelemetry.io/otel/attribute"
 )
@@ -31,6 +33,7 @@ func (app *App) CheckTx(ctx context.Context, req *abci.RequestCheckTx) (*abci.Re
 }
 
 func (app *App) DeliverTx(ctx sdk.Context, req abci.RequestDeliverTx) abci.ResponseDeliverTx {
+	defer metrics.MeasureDeliverTxDuration(time.Now())
 	// tracectx, span := (*app.tracingInfo.Tracer).Start(app.tracingInfo.TracerContext, "DeliverTx")
 	// oldCtx := app.tracingInfo.TracerContext
 	// app.tracingInfo.TracerContext = tracectx

app/app.go

@@ -108,6 +108,7 @@ import (
 	tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
 	dbm "github.com/tendermint/tm-db"
 
+	"github.com/sei-protocol/sei-chain/utils/metrics"
 	"github.com/sei-protocol/sei-chain/utils/tracing"
 
 	dexmodule "github.com/sei-protocol/sei-chain/x/dex"
@@ -950,6 +951,7 @@ func (app *App) ProcessBlockSynchronous(ctx sdk.Context, txs [][]byte) []*abci.E
 	for _, tx := range txs {
 		txResults = append(txResults, app.DeliverTxWithResult(ctx, tx))
 	}
+	metrics.IncrTxProcessTypeCounter(metrics.SYNCHRONOUS)
 	return txResults
 }
 
@@ -989,6 +991,7 @@ func (app *App) ProcessTxConcurrent(
 
 	// Deliver the transaction and store the result in the channel
 	resultChan <- ChannelResult{txIndex, app.DeliverTxWithResult(ctx, txBytes)}
+	metrics.IncrTxProcessTypeCounter(metrics.CONCURRENT)
 }
 
 func (app *App) ProcessBlockConcurrent(
@@ -1020,10 +1023,15 @@ func (app *App) ProcessBlockConcurrent(
 		)
 	}
 
-	// Waits for all the transactions to complete
-	waitGroup.Wait()
+	// Do not call waitGroup.Wait() synchronously as it blocks on channel reads
+	// until all the messages are read. This closes the channel once
+	// results are all read and prevent any further writes.
+	go func() {
+		waitGroup.Wait()
+		close(resultChan)
+	}()
 
-	// Gather Results and store it based on txIndex
+	// Gather Results and store it based on txIndex and read results from channel
 	// Concurrent results may be in different order than the original txIndex
 	txResultsMap := map[int]*abci.ExecTxResult{}
 	for result := range resultChan {

app/graph.go

@@ -2,9 +2,11 @@ package app
 
 import (
 	"fmt"
+	"time"
 
 	acltypes "github.com/cosmos/cosmos-sdk/types/accesscontrol"
 	mapset "github.com/deckarep/golang-set"
+	"github.com/sei-protocol/sei-chain/utils/metrics"
 )
 
 type DagNodeID int
@@ -133,6 +135,7 @@ func (dag *Dag) AddEdge(fromIndex DagNodeID, toIndex DagNodeID) *DagEdge {
 //
 // It will also register the new node with AccessOpsMap so that future nodes that amy be dependent on this one can properly identify the dependency.
 func (dag *Dag) AddNodeBuildDependency(messageIndex int, txIndex int, accessOp acltypes.AccessOperation) {
+	defer metrics.MeasureBuildDagDuration(time.Now(), "AddNodeBuildDependency")
 	dagNode := dag.AddNode(messageIndex, txIndex, accessOp)
 	// update tx index map
 	dag.TxIndexMap[txIndex] = dagNode.NodeID
@@ -283,6 +286,7 @@ func (dag *Dag) BuildCompletionSignalMaps() (
 	completionSignalingMap map[int]MessageCompletionSignalMapping,
 	blockingSignalsMap map[int]MessageCompletionSignalMapping,
 ) {
+	defer metrics.MeasureBuildDagDuration(time.Now(), "BuildCompletionSignalMaps")
 	completionSignalingMap = make(map[int]MessageCompletionSignalMapping)
 	blockingSignalsMap = make(map[int]MessageCompletionSignalMapping)
 	// go through every node

go.mod

@@ -131,7 +131,7 @@ require (
 )
 
 replace (
-	github.com/cosmos/cosmos-sdk => github.com/sei-protocol/sei-cosmos v0.1.73
+	github.com/cosmos/cosmos-sdk => github.com/sei-protocol/sei-cosmos v0.1.76
 	github.com/gogo/protobuf => github.com/regen-network/protobuf v1.3.3-alpha.regen.1
 	github.com/keybase/go-keychain => github.com/99designs/go-keychain v0.0.0-20191008050251-8e49817e8af4
 	github.com/tendermint/tendermint => github.com/sei-protocol/sei-tendermint v0.1.56

go.sum

@@ -1097,8 +1097,8 @@ github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg
 github.com/seccomp/libseccomp-golang v0.9.1/go.mod h1:GbW5+tmTXfcxTToHLXlScSlAvWlF4P2Ca7zGrPiEpWo=
 github.com/securego/gosec/v2 v2.11.0/go.mod h1:SX8bptShuG8reGC0XS09+a4H2BoWSJi+fscA+Pulbpo=
 github.com/segmentio/fasthash v1.0.3/go.mod h1:waKX8l2N8yckOgmSsXJi7x1ZfdKZ4x7KRMzBtS3oedY=
-github.com/sei-protocol/sei-cosmos v0.1.73 h1:yFG32zV8BGCbsS/AQYUNc2ZFWWZ2lXv505Gkf6vauKE=
-github.com/sei-protocol/sei-cosmos v0.1.73/go.mod h1:Oaj7toqHCkwEEb+sDIWxtfTkPZxOpMXBXDMvIIqUjpw=
+github.com/sei-protocol/sei-cosmos v0.1.76 h1:y75Pn8hTc1SWndq/oMGzVu4mkF9ma6kVaA36AdFKxD8=
+github.com/sei-protocol/sei-cosmos v0.1.76/go.mod h1:Oaj7toqHCkwEEb+sDIWxtfTkPZxOpMXBXDMvIIqUjpw=
 github.com/sei-protocol/sei-tendermint v0.1.56 h1:iRVhiIetj+GSwpBzaR9lqgvmgcOqmShfACS6x3tuBIw=
 github.com/sei-protocol/sei-tendermint v0.1.56/go.mod h1:Olwbjyagrpoxj5DAUhHxMTWDVEfQ3FYdpypaJ3+6Hs8=
 github.com/sergi/go-diff v1.1.0/go.mod h1:STckp+ISIX8hZLjrqAeVduY0gWCT9IjLuqbuNXdaHfM=

loadtest/contracts/deploy_ten_contracts.sh

@@ -15,17 +15,17 @@ echo
 cd $seihome/loadtest/contracts/mars && cargo build && docker run --rm -v "$(pwd)":/code \
   --mount type=volume,source="$(basename "$(pwd)")_cache",target=/code/target \
   --mount type=volume,source=registry_cache,target=/usr/local/cargo/registry \
-  cosmwasm/rust-optimizer:0.12.5
+  cosmwasm/rust-optimizer:0.12.6
 
 cd $seihome/loadtest/contracts/saturn && cargo build && docker run --rm -v "$(pwd)":/code \
   --mount type=volume,source="$(basename "$(pwd)")_cache",target=/code/target \
   --mount type=volume,source=registry_cache,target=/usr/local/cargo/registry \
-  cosmwasm/rust-optimizer:0.12.5
+  cosmwasm/rust-optimizer:0.12.6
 
 cd $seihome/loadtest/contracts/venus && cargo build && docker run --rm -v "$(pwd)":/code \
   --mount type=volume,source="$(basename "$(pwd)")_cache",target=/code/target \
   --mount type=volume,source=registry_cache,target=/usr/local/cargo/registry \
-  cosmwasm/rust-optimizer:0.12.5
+  cosmwasm/rust-optimizer:0.12.6
 
 # Deploy all contracts
 echo "Deploying contracts..."
@@ -70,7 +70,7 @@ marsaddr4=$(python3 parser.py contract_address $marsinsres4)
 echo "Registering..."
 
 valaddr=$(printf "12345678\n" | $seidbin keys show $(printf "12345678\n" | $seidbin keys show node_admin --output json | jq -r .address) --bech=val --output json | jq -r '.address')
-printf "12345678\n" | $seidbin tx staking delegate $valaddr 1000000000usei --from=$keyname --chain-id=$chainid -b block -y
+printf "12345678\n" | $seidbin tx staking delegate $valaddr 1000000000usei --from=$keyname --chain-id=$chainid -b block -y --fees 2000usei
 
 printf "12345678\n" | $seidbin tx dex register-contract $marsaddr $marsid false true  -y --from=$keyname --chain-id=$chainid --fees=10000000usei --gas=500000 --broadcast-mode=block
 printf "12345678\n" | $seidbin tx dex register-contract $saturnaddr $saturnid false true -y --from=$keyname --chain-id=$chainid --fees=10000000usei --gas=500000 --broadcast-mode=block
@@ -145,7 +145,7 @@ printf "12345678\n" | $seidbin tx gov vote $marsproposalid4 yes -y --from=$keyna
 
 sleep 90
 
-printf "12345678\n" | $seidbin tx staking unbond $valaddr 1000000000usei --from=$keyname --chain-id=$chainid -b block -y
+printf "12345678\n" | $seidbin tx staking unbond $valaddr 1000000000usei --from=$keyname --chain-id=$chainid -b block -y --fees 2000usei
 
 echo $marsaddr
 echo $saturnaddr

utils/metrics/labels.go

@@ -0,0 +1,6 @@
+package metrics
+
+const (
+	CONCURRENT  = "concurrent"
+	SYNCHRONOUS = "synchronous"
+)

utils/metrics/metrics_util.go

@@ -0,0 +1,47 @@
+package metrics
+
+import (
+	"time"
+
+	metrics "github.com/armon/go-metrics"
+	"github.com/cosmos/cosmos-sdk/telemetry"
+)
+
+// Measures the time taken to execute a sudo msg
+// Metric Names:
+//
+//	sei_tx_process_type_count
+func IncrTxProcessTypeCounter(processType string) {
+	metrics.IncrCounterWithLabels(
+		[]string{"sei", "tx", "process", "type"},
+		1,
+		[]metrics.Label{telemetry.NewLabel("type", processType)},
+	)
+}
+
+// Measures the time taken to execute a sudo msg
+// Metric Names:
+//
+//	sei_deliver_tx_duration_miliseconds
+//	sei_deliver_tx_duration_miliseconds_count
+//	sei_deliver_tx_duration_miliseconds_sum
+func MeasureDeliverTxDuration(start time.Time) {
+	metrics.MeasureSince(
+		[]string{"sei", "deliver", "tx", "milliseconds"},
+		start.UTC(),
+	)
+}
+
+// Measures the time taken to execute a sudo msg
+// Metric Names:
+//
+//	sei_dag_build_duration_miliseconds
+//	sei_dag_build_duration_miliseconds_count
+//	sei_dag_build_duration_miliseconds_sum
+func MeasureBuildDagDuration(start time.Time, method string) {
+	metrics.MeasureSinceWithLabels(
+		[]string{"sei", "dag", "build", "milliseconds"},
+		start.UTC(),
+		[]metrics.Label{telemetry.NewLabel("method", method)},
+	)
+}

x/dex/keeper/epoch.go

@@ -14,13 +14,13 @@ func (k Keeper) SetEpoch(ctx sdk.Context, epoch uint64) {
 	bz := make([]byte, 8)
 	binary.BigEndian.PutUint64(bz, epoch)
 	store.Set([]byte(EpochKey), bz)
+	ctx.Logger().Info(fmt.Sprintf("Current epoch %d", epoch))
 }
 
 func (k Keeper) IsNewEpoch(ctx sdk.Context) (bool, uint64) {
 	store := ctx.KVStore(k.storeKey)
 	b := store.Get([]byte(EpochKey))
 	lastEpoch := binary.BigEndian.Uint64(b)
 	currentEpoch := k.EpochKeeper.GetEpoch(ctx).CurrentEpoch
-	ctx.Logger().Info(fmt.Sprintf("Current epoch %d", currentEpoch))
 	return currentEpoch > lastEpoch, currentEpoch
 }

x/oracle/simulation/operations.go

@@ -83,7 +83,7 @@ func WeightedOperations(
 }
 
 // SimulateMsgAggregateExchangeRatePrevote generates a MsgAggregateExchangeRatePrevote with random values.
-// nolint: funlen
+//nolint: funlen
 func SimulateMsgAggregateExchangeRatePrevote(ak types.AccountKeeper, bk types.BankKeeper, k keeper.Keeper) simtypes.Operation {
 	return func(
 		r *rand.Rand, app *baseapp.BaseApp, ctx sdk.Context, accs []simtypes.Account, chainID string,
@@ -146,7 +146,7 @@ func SimulateMsgAggregateExchangeRatePrevote(ak types.AccountKeeper, bk types.Ba
 }
 
 // SimulateMsgAggregateExchangeRateVote generates a MsgAggregateExchangeRateVote with random values.
-// nolint: funlen
+//nolint: funlen
 func SimulateMsgAggregateExchangeRateVote(ak types.AccountKeeper, bk types.BankKeeper, k keeper.Keeper) simtypes.Operation {
 	return func(
 		r *rand.Rand, app *baseapp.BaseApp, ctx sdk.Context, accs []simtypes.Account, chainID string,
@@ -213,7 +213,7 @@ func SimulateMsgAggregateExchangeRateVote(ak types.AccountKeeper, bk types.BankK
 }
 
 // SimulateMsgDelegateFeedConsent generates a MsgDelegateFeedConsent with random values.
-// nolint: funlen
+//nolint: funlen
 func SimulateMsgDelegateFeedConsent(ak types.AccountKeeper, bk types.BankKeeper, k keeper.Keeper) simtypes.Operation {
 	return func(
 		r *rand.Rand, app *baseapp.BaseApp, ctx sdk.Context, accs []simtypes.Account, chainID string,