Bweng acl order fix

app/abci.go

@@ -34,11 +34,11 @@ 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
-	// defer span.End()
-	// defer func() { app.tracingInfo.TracerContext = oldCtx }()
+	tracectx, span := (*app.tracingInfo.Tracer).Start(app.tracingInfo.TracerContext, "DeliverTx")
+	oldCtx := app.tracingInfo.TracerContext
+	app.tracingInfo.TracerContext = tracectx
+	defer span.End()
+	defer func() { app.tracingInfo.TracerContext = oldCtx }()
 	return app.BaseApp.DeliverTx(ctx, req)
 }
 

app/app.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	storetypes "github.com/cosmos/cosmos-sdk/store/types"
+	"github.com/k0kubun/pp"
 
 	"github.com/sei-protocol/sei-chain/aclmapping"
 	appparams "github.com/sei-protocol/sei-chain/app/params"
@@ -979,8 +980,18 @@ func (app *App) ProcessTxConcurrent(
 	ctx = ctx.WithTxCompletionChannels(getChannelsFromSignalMapping(txCompletionSignalingMap))
 
 	// Deliver the transaction and store the result in the channel
+	ctx.Logger().Info(fmt.Sprintf("ProcessTxConcurrent::Sent Tx Index=%d", txIndex))
 	resultChan <- ChannelResult{txIndex, app.DeliverTxWithResult(ctx, txBytes)}
 	metrics.IncrTxProcessTypeCounter(metrics.CONCURRENT)
+	ctx.Logger().Info(fmt.Sprintf("ProcessTxConcurrent::Processed Tx Index=%d", txIndex))
+}
+
+// cacheContext returns a new context based off of the provided context with
+// a branched multi-store.
+func (app *App) CacheContext(ctx sdk.Context) (sdk.Context, sdk.CacheMultiStore) {
+	ms := ctx.MultiStore()
+	msCache := ms.CacheMultiStore()
+	return ctx.WithMultiStore(msCache), msCache
 }
 
 func (app *App) ProcessBlockConcurrent(
@@ -1001,6 +1012,7 @@ func (app *App) ProcessBlockConcurrent(
 	}
 
 	// For each transaction, start goroutine and deliver TX
+	ctx.Logger().Info(fmt.Sprintf("ProcessBlock: sending off num_txs=%d", len(txs)))
 	for txIndex, txBytes := range txs {
 		waitGroup.Add(1)
 		go app.ProcessTxConcurrent(
@@ -1024,11 +1036,12 @@ func (app *App) ProcessBlockConcurrent(
 
 	// Gather Results and store it based on txIndex and read results from channel
 	// Concurrent results may be in different order than the original txIndex
+	ctx.Logger().Info("ProcessBlock:collecting transactions")
 	txResultsMap := map[int]*abci.ExecTxResult{}
 	for result := range resultChan {
 		txResultsMap[result.txIndex] = result.result
 	}
-
+	ctx.Logger().Info("ProcessBlock:collected transactions")
 	// Gather Results and store in array based on txIndex to preserve ordering
 	for txIndex := range txs {
 		txResults = append(txResults, txResultsMap[txIndex])
@@ -1079,12 +1092,24 @@ func (app *App) ProcessBlock(ctx sdk.Context, txs [][]byte, req BlockProcessRequ
 	// app.batchVerifier.VerifyTxs(ctx, typedTxs)
 
 	dependencyDag, err := app.AccessControlKeeper.BuildDependencyDag(ctx, app.txDecoder, app.GetAnteDepGenerator(), txs)
+	pp.Printf("ProcessBlock:: Dependency DAG:%s", dependencyDag)
+
 	var txResults []*abci.ExecTxResult
 
 	switch err {
 	case nil:
 		// Only run concurrently if no error
-		txResults = app.ProcessBlockConcurrent(ctx, txs, dependencyDag.CompletionSignalingMap, dependencyDag.BlockingSignalsMap)
+
+		// Branch off the current context and pass a cached context to the concurrent delivered TXs that are shared.
+		// runTx will write to this ephermeral CacheMultiStore, after the process block is done, Write() is called on this
+		// CacheMultiStore where it writes the data to the parent store (DeliverState) in sorted Key order to maintain
+		// deterministic ordering between validators in the case of concurrent deliverTXs
+		processBlockCtx, processBlockCache := app.CacheContext(ctx)
+		txResults = app.ProcessBlockConcurrent(processBlockCtx, txs, dependencyDag.CompletionSignalingMap, dependencyDag.BlockingSignalsMap)
+		// Write the results back to the concurrent contexts
+		ctx.Logger().Info("ProcessBlock:Writing processBlockCtx")
+		processBlockCache.Write()
+		ctx.Logger().Info("ProcessBlock:processBlockCtx wrote")
 	case acltypes.ErrGovMsgInBlock:
 		ctx.Logger().Info(fmt.Sprintf("Gov msg found while building DAG, processing synchronously: %s", err))
 		txResults = app.ProcessBlockSynchronous(ctx, txs)

go.mod

@@ -13,6 +13,7 @@ require (
 	github.com/golang/protobuf v1.5.2
 	github.com/gorilla/mux v1.8.0
 	github.com/grpc-ecosystem/grpc-gateway v1.16.0
+	github.com/k0kubun/pp v3.0.1+incompatible
 	github.com/pkg/errors v0.9.1
 	github.com/regen-network/cosmos-proto v0.3.1
 	github.com/spf13/cast v1.5.0
@@ -88,6 +89,7 @@ require (
 	github.com/lib/pq v1.10.6 // indirect
 	github.com/libp2p/go-buffer-pool v0.0.2 // indirect
 	github.com/magiconair/properties v1.8.6 // indirect
+	github.com/mattn/go-colorable v0.1.12 // indirect
 	github.com/mattn/go-isatty v0.0.14 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 // indirect
 	github.com/mitchellh/go-homedir v1.1.0 // indirect
@@ -131,7 +133,7 @@ require (
 )
 
 replace (
-	github.com/cosmos/cosmos-sdk => github.com/sei-protocol/sei-cosmos v0.1.114
+	github.com/cosmos/cosmos-sdk => github.com/sei-protocol/sei-cosmos v0.1.150
 	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.59

go.sum

@@ -733,7 +733,10 @@ github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7V
 github.com/julienschmidt/httprouter v1.3.0/go.mod h1:JR6WtHb+2LUe8TCKY3cZOxFyyO8IZAc4RVcycCCAKdM=
 github.com/julz/importas v0.1.0/go.mod h1:oSFU2R4XK/P7kNBrnL/FEQlDGN1/6WoxXEjSSXO0DV0=
 github.com/jung-kurt/gofpdf v1.0.3-0.20190309125859-24315acbbda5/go.mod h1:7Id9E/uU8ce6rXgefFLlgrJj/GYY22cpxn+r32jIOes=
+github.com/k0kubun/colorstring v0.0.0-20150214042306-9440f1994b88 h1:uC1QfSlInpQF+M0ao65imhwqKnz3Q2z/d8PWZRMQvDM=
 github.com/k0kubun/colorstring v0.0.0-20150214042306-9440f1994b88/go.mod h1:3w7q1U84EfirKl04SVQ/s7nPm1ZPhiXd34z40TNz36k=
+github.com/k0kubun/pp v3.0.1+incompatible h1:3tqvf7QgUnZ5tXO6pNAZlrvHgl6DvifjDrd9g2S9Z40=
+github.com/k0kubun/pp v3.0.1+incompatible/go.mod h1:GWse8YhT0p8pT4ir3ZgBbfZild3tgzSScAn6HmfYukg=
 github.com/karalabe/usb v0.0.0-20190919080040-51dc0efba356/go.mod h1:Od972xHfMJowv7NGVDiWVxk2zxnWgjLlJzE+F4F7AGU=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/errcheck v1.6.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
@@ -799,6 +802,7 @@ github.com/mattn/go-colorable v0.1.4/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVc
 github.com/mattn/go-colorable v0.1.6/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
 github.com/mattn/go-colorable v0.1.8/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
 github.com/mattn/go-colorable v0.1.9/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
+github.com/mattn/go-colorable v0.1.12 h1:jF+Du6AlPIjs2BiUiQlKOX0rt3SujHxPnksPKZbaA40=
 github.com/mattn/go-colorable v0.1.12/go.mod h1:u5H1YNBxpqRaxsYJYSkiCWKzEfiAb1Gb520KVy5xxl4=
 github.com/mattn/go-ieproxy v0.0.0-20190610004146-91bb50d98149/go.mod h1:31jz6HNzdxOmlERGGEc4v/dMssOfmp2p5bT/okiKFFc=
 github.com/mattn/go-ieproxy v0.0.0-20190702010315-6dee0af9227d/go.mod h1:31jz6HNzdxOmlERGGEc4v/dMssOfmp2p5bT/okiKFFc=
@@ -1097,8 +1101,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.114 h1:Cy5MnBdvql5VJw5pC104DsivQxPg0xkvVOuq6VwDiRk=
-github.com/sei-protocol/sei-cosmos v0.1.114/go.mod h1:8ccWQxpBkWbpvBos/T4QO9K9gQxFs0duTqKRnagKo+0=
+github.com/sei-protocol/sei-cosmos v0.1.150 h1:RSdQ2nunKz1lK3Gdl8K/HE4qQWGvFBddBozQ4yBiemI=
+github.com/sei-protocol/sei-cosmos v0.1.150/go.mod h1:8ccWQxpBkWbpvBos/T4QO9K9gQxFs0duTqKRnagKo+0=
 github.com/sei-protocol/sei-tendermint v0.1.59 h1:POGL60PumMQHF4EzAHzvkGfDnodQJLHpl65LuiwSO/Y=
 github.com/sei-protocol/sei-tendermint v0.1.59/go.mod h1:Olwbjyagrpoxj5DAUhHxMTWDVEfQ3FYdpypaJ3+6Hs8=
 github.com/sergi/go-diff v1.1.0/go.mod h1:STckp+ISIX8hZLjrqAeVduY0gWCT9IjLuqbuNXdaHfM=

loadtest/config.json

@@ -1,8 +1,8 @@
 {
     "batch_size": 10,
     "chain_id": "sei-loadtest-testnet",
-    "orders_per_block": 400,
-    "rounds": 5,
+    "orders_per_block": 40,
+    "rounds": 1,
     "price_distribution": {
         "min": "45",
         "max": "55",

loadtest/main.go

@@ -169,7 +169,7 @@ func run(config Config) {
 		wg := &sync.WaitGroup{}
 		var senders []func()
 		wgs = append(wgs, wg)
-		for _, account := range activeAccounts {
+		for i, account := range activeAccounts {
 			key := GetKey(uint64(account))
 
 			msg := generateMessage(config, key, batchSize)