Ezhu/dex parallelization

aclmapping/dex/mappings.go

@@ -7,17 +7,20 @@ import (
 	sdkacltypes "github.com/cosmos/cosmos-sdk/types/accesscontrol"
 	aclkeeper "github.com/cosmos/cosmos-sdk/x/accesscontrol/keeper"
 	acltypes "github.com/cosmos/cosmos-sdk/x/accesscontrol/types"
+	utils "github.com/sei-protocol/sei-chain/aclmapping/utils"
 	dexmoduletypes "github.com/sei-protocol/sei-chain/x/dex/types"
 )
 
-var ErrPlaceOrdersGenerator = fmt.Errorf("invalid message received for type DexPlaceOrders")
+var ErrPlaceOrdersGenerator = fmt.Errorf("invalid message received for dex module")
 
 func GetDexDependencyGenerators() aclkeeper.DependencyGeneratorMap {
 	dependencyGeneratorMap := make(aclkeeper.DependencyGeneratorMap)
 
 	// dex place orders
 	placeOrdersKey := acltypes.GenerateMessageKey(&dexmoduletypes.MsgPlaceOrders{})
+	cancelOrdersKey := acltypes.GenerateMessageKey(&dexmoduletypes.MsgCancelOrders{})
 	dependencyGeneratorMap[placeOrdersKey] = DexPlaceOrdersDependencyGenerator
+	dependencyGeneratorMap[cancelOrdersKey] = DexCancelOrdersDependencyGenerator
 
 	return dependencyGeneratorMap
 }
@@ -27,8 +30,51 @@ func DexPlaceOrdersDependencyGenerator(keeper aclkeeper.Keeper, ctx sdk.Context,
 	if !ok {
 		return []sdkacltypes.AccessOperation{}, ErrPlaceOrdersGenerator
 	}
-	// TODO: This is not final, JUST AN EXAMPLE
+
 	return []sdkacltypes.AccessOperation{
-		{AccessType: sdkacltypes.AccessType_WRITE, ResourceType: sdkacltypes.ResourceType_KV, IdentifierTemplate: placeOrdersMsg.ContractAddr},
+		{
+			AccessType:         sdkacltypes.AccessType_READ,
+			ResourceType:       sdkacltypes.ResourceType_KV_DEX,
+			IdentifierTemplate: utils.GetIdentifierTemplatePerModule(utils.DEX, placeOrdersMsg.ContractAddr),
+		},
+		{
+			AccessType:         sdkacltypes.AccessType_WRITE,
+			ResourceType:       sdkacltypes.ResourceType_KV_DEX,
+			IdentifierTemplate: utils.GetIdentifierTemplatePerModule(utils.DEX, placeOrdersMsg.ContractAddr),
+		},
+
+		// Last Operation should always be a commit
+		{
+			ResourceType:       sdkacltypes.ResourceType_ANY,
+			AccessType:         sdkacltypes.AccessType_COMMIT,
+			IdentifierTemplate: utils.DefaultIDTemplate,
+		},
+	}, nil
+}
+
+func DexCancelOrdersDependencyGenerator(keeper aclkeeper.Keeper, ctx sdk.Context, msg sdk.Msg) ([]sdkacltypes.AccessOperation, error) {
+	cancelOrdersMsg, ok := msg.(*dexmoduletypes.MsgCancelOrders)
+	if !ok {
+		return []sdkacltypes.AccessOperation{}, ErrPlaceOrdersGenerator
+	}
+
+	return []sdkacltypes.AccessOperation{
+		{
+			AccessType:         sdkacltypes.AccessType_READ,
+			ResourceType:       sdkacltypes.ResourceType_KV_DEX,
+			IdentifierTemplate: utils.GetIdentifierTemplatePerModule(utils.DEX, cancelOrdersMsg.ContractAddr),
+		},
+		{
+			AccessType:         sdkacltypes.AccessType_WRITE,
+			ResourceType:       sdkacltypes.ResourceType_KV_DEX,
+			IdentifierTemplate: utils.GetIdentifierTemplatePerModule(utils.DEX, cancelOrdersMsg.ContractAddr),
+		},
+
+		// Last Operation should always be a commit
+		{
+			ResourceType:       sdkacltypes.ResourceType_ANY,
+			AccessType:         sdkacltypes.AccessType_COMMIT,
+			IdentifierTemplate: utils.DefaultIDTemplate,
+		},
 	}, nil
 }

aclmapping/dex/mappings_test.go

@@ -0,0 +1 @@
+package acldexmapping

aclmapping/utils/identifier_templates.go

@@ -13,6 +13,7 @@ const (
 	AUTH              = "auth"
 	STAKING           = "staking"
 	TOKENFACTORY      = "tokenfactory"
+	DEX               = "dex"
 	DefaultIDTemplate = "*"
 )
 

app/app.go

@@ -1009,7 +1009,6 @@ func (app *App) ProcessTxConcurrent(
 	ctx = ctx.WithTxMsgAccessOps(txMsgAccessOpMapping)
 
 	// Deliver the transaction and store the result in the channel
-
 	resultChan <- ChannelResult{txIndex, app.DeliverTxWithResult(ctx, txBytes)}
 	metrics.IncrTxProcessTypeCounter(metrics.CONCURRENT)
 }

loadtest/config.json

@@ -15,32 +15,36 @@
     },
     "message_type_distribution": {
         "dex": {
-            "limit_order_percentage": "0.2",
-            "market_order_percentage": "0.8"
+            "place_order_percentage": "0.5",
+            "cancel_order_percentage": "0.5"
         },
         "staking": {
             "delegate_percentage": "0.5",
             "undelegate_percentage": "0.25",
             "begin_redelegate_percentage": "0.25"
         }
     },
-    "run_oracle": false,
-    "message_type": "basic",
+    "order_type_distribution": {
+        "limit_order_percentage": "0.2",
+        "market_order_percentage": "0.8"
+    },
+    "message_type": "dex",
+    "run_oracle": true,
     "contract_distribution": [
         {
-            "contract_address": "sei1yw4xvtc43me9scqfr2jr2gzvcxd3a9y4eq7gaukreugw2yd2f8tsy4qgdm",
+            "contract_address": "sei14hj2tavq8fpesdwxxcu44rty3hh90vhujrvcmstl4zr3txmfvw9sh9m79m",
             "percentage": "0.25"
         },
         {
-            "contract_address": "sei1wug8sewp6cedgkmrmvhl3lf3tulagm9hnvy8p0rppz9yjw0g4wtq05fpn3",
+            "contract_address": "sei1nc5tatafv6eyq7llkr2gv50ff9e22mnf70qgjlv737ktmt4eswrqms7u8a",
             "percentage": "0.25"
         },
         {
-            "contract_address": "sei1qg5ega6dykkxc307y25pecuufrjkxkaggkkxh7nad0vhyhtuhw3scwfwpd",
+            "contract_address": "sei17p9rzwnnfxcjp32un9ug7yhhzgtkhvl9jfksztgw5uh69wac2pgsrtqewe",
             "percentage": "0.25"
         },
         {
-            "contract_address": "sei1zwv6feuzhy6a9wekh96cd57lsarmqlwxdypdsplw6zhfncqw6ftqr428wx",
+            "contract_address": "sei1yw4xvtc43me9scqfr2jr2gzvcxd3a9y4eq7gaukreugw2yd2f8tsy4qgdm",
             "percentage": "0.25"
         }
     ]

loadtest/loadtest_client.go

@@ -10,13 +10,15 @@ import (
 
 	cryptotypes "github.com/cosmos/cosmos-sdk/crypto/types"
 	typestx "github.com/cosmos/cosmos-sdk/types/tx"
+	dextypes "github.com/sei-protocol/sei-chain/x/dex/types"
 	"google.golang.org/grpc"
 )
 
 type LoadTestClient struct {
 	LoadTestConfig  Config
 	TestConfig      EncodingConfig
 	TxClient        typestx.ServiceClient
+	DexQueryClient  dextypes.QueryClient
 	SignerClient    *SignerClient
 	ChainID         string
 	TxHashList      []string
@@ -30,7 +32,7 @@ func NewLoadTestClient() *LoadTestClient {
 		grpc.WithInsecure(),
 	)
 	TxClient := typestx.NewServiceClient(grpcConn)
-
+	DexQueryClient := dextypes.NewQueryClient(grpcConn)
 	config := Config{}
 	pwd, _ := os.Getwd()
 	file, _ := os.ReadFile(pwd + "/loadtest/config.json")
@@ -42,6 +44,7 @@ func NewLoadTestClient() *LoadTestClient {
 		LoadTestConfig:  config,
 		TestConfig:      TestConfig,
 		TxClient:        TxClient,
+		DexQueryClient:  DexQueryClient,
 		SignerClient:    NewSignerClient(),
 		ChainID:         config.ChainID,
 		TxHashList:      []string{},
@@ -99,15 +102,17 @@ func (c *LoadTestClient) BuildTxs() (workgroups []*sync.WaitGroup, sendersList [
 		var senders []func()
 		workgroups = append(workgroups, wg)
 		if config.MessageType != "none" {
-			for _, account := range activeAccounts {
+			for j, account := range activeAccounts {
 				key := c.SignerClient.GetKey(uint64(account))
 
-				msg := generateMessage(config, key, config.MsgsPerTx, qv.Validators)
+				msg := generateMessage(c, key, config.MsgsPerTx, qv.Validators)
 				txBuilder := TestConfig.TxConfig.NewTxBuilder()
 				_ = txBuilder.SetMsgs(msg)
 				seqDelta := uint64(i / 2)
 				mode := typestx.BroadcastMode_BROADCAST_MODE_SYNC
-
+				if j == len(activeAccounts)-1 {
+					mode = typestx.BroadcastMode_BROADCAST_MODE_BLOCK
+				}
 				// Note: There is a potential race condition here with seqnos
 				// in which a later seqno is delievered before an earlier seqno
 				// In practice, we haven't run into this issue so we'll leave this

loadtest/main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"math/rand"
@@ -71,9 +72,9 @@ func run() {
 	fmt.Printf("%s - Finished\n", time.Now().Format("2006-01-02T15:04:05"))
 }
 
-func generateMessage(config Config, key cryptotypes.PrivKey, msgPerTx uint64, validators []Validator) sdk.Msg {
+func generateMessage(c *LoadTestClient, key cryptotypes.PrivKey, msgPerTx uint64, validators []Validator) sdk.Msg {
 	var msg sdk.Msg
-	switch config.MessageType {
+	switch c.LoadTestConfig.MessageType {
 	case "basic":
 		msg = &banktypes.MsgSend{
 			FromAddress: sdk.AccAddress(key.PubKey().Address()).String(),
@@ -84,7 +85,7 @@ func generateMessage(config Config, key cryptotypes.PrivKey, msgPerTx uint64, va
 			}),
 		}
 	case "staking":
-		msgType := config.MsgTypeDistr.SampleStakingMsgs()
+		msgType := c.LoadTestConfig.MsgTypeDistr.SampleStakingMsgs()
 
 		switch msgType {
 		case "delegate":
@@ -110,48 +111,90 @@ func generateMessage(config Config, key cryptotypes.PrivKey, msgPerTx uint64, va
 			panic("Unknown message type")
 		}
 	case "dex":
-		msgType := config.MsgTypeDistr.SampleDexMsgs()
-		orderPlacements := []*dextypes.Order{}
-		var orderType dextypes.OrderType
-		if msgType == "limit" {
-			orderType = dextypes.OrderType_LIMIT
-		} else {
-			orderType = dextypes.OrderType_MARKET
-		}
-		var direction dextypes.PositionDirection
-		if rand.Float64() < 0.5 {
-			direction = dextypes.PositionDirection_LONG
-		} else {
-			direction = dextypes.PositionDirection_SHORT
-		}
-		price := config.PriceDistr.Sample()
-		quantity := config.QuantityDistr.Sample()
-		contract := config.ContractDistr.Sample()
-		for j := 0; j < int(msgPerTx); j++ {
-			orderPlacements = append(orderPlacements, &dextypes.Order{
-				Account:           sdk.AccAddress(key.PubKey().Address()).String(),
-				ContractAddr:      contract,
-				PositionDirection: direction,
-				Price:             price.Quo(FromMili),
-				Quantity:          quantity.Quo(FromMili),
-				PriceDenom:        "SEI",
-				AssetDenom:        "ATOM",
-				OrderType:         orderType,
-				Data:              VortexData,
-			})
-		}
-		amount, err := sdk.ParseCoinsNormalized(fmt.Sprintf("%d%s", price.Mul(quantity).Ceil().RoundInt64(), "usei"))
-		if err != nil {
-			panic(err)
-		}
-		msg = &dextypes.MsgPlaceOrders{
-			Creator:      sdk.AccAddress(key.PubKey().Address()).String(),
-			Orders:       orderPlacements,
-			ContractAddr: contract,
-			Funds:        amount,
+		msgType := c.LoadTestConfig.MsgTypeDistr.SampleDexMsgs()
+		switch msgType {
+		case "place_order":
+			orderPlacements := []*dextypes.Order{}
+			var direction dextypes.PositionDirection
+			if rand.Float64() < 0.5 {
+				direction = dextypes.PositionDirection_LONG
+			} else {
+				direction = dextypes.PositionDirection_SHORT
+			}
+			orderType := c.LoadTestConfig.OrderTypeDistr.SampleOrderType()
+			price := c.LoadTestConfig.PriceDistr.Sample()
+			quantity := c.LoadTestConfig.QuantityDistr.Sample()
+			contract := c.LoadTestConfig.ContractDistr.Sample()
+			for j := 0; j < int(msgPerTx); j++ {
+				order := &dextypes.Order{
+					Account:           sdk.AccAddress(key.PubKey().Address()).String(),
+					ContractAddr:      contract,
+					PositionDirection: direction,
+					Price:             price.Quo(FromMili),
+					Quantity:          quantity.Quo(FromMili),
+					PriceDenom:        "SEI",
+					AssetDenom:        "ATOM",
+					OrderType:         orderType,
+					Data:              VortexData,
+				}
+				orderPlacements = append(orderPlacements, order)
+			}
+			amount, err := sdk.ParseCoinsNormalized(fmt.Sprintf("%d%s", price.Mul(quantity).Ceil().RoundInt64(), "usei"))
+			if err != nil {
+				panic(err)
+			}
+			msg = &dextypes.MsgPlaceOrders{
+				Creator:      sdk.AccAddress(key.PubKey().Address()).String(),
+				Orders:       orderPlacements,
+				ContractAddr: contract,
+				Funds:        amount,
+			}
+		case "cancel_order":
+			var contract string
+			var outstandingOrders []*dextypes.Order
+			for _, contractConfig := range c.LoadTestConfig.ContractDistr {
+
+				if resp, err := c.DexQueryClient.GetOrders(context.Background(), &dextypes.QueryGetOrdersRequest{
+					ContractAddr: contractConfig.ContractAddr,
+					Account:      sdk.AccAddress(key.PubKey().Address()).String(),
+				}); err != nil {
+					panic(err)
+				} else if len(resp.Orders) > 0 {
+					contract = contractConfig.ContractAddr
+					outstandingOrders = resp.Orders
+					break
+				}
+			}
+
+			cancelPlacements := []*dextypes.Cancellation{}
+			for j := 0; j < int(msgPerTx); j++ {
+				if len(outstandingOrders) > 0 {
+					order := outstandingOrders[len(outstandingOrders)-1]
+					outstandingOrders = outstandingOrders[:len(outstandingOrders)-1]
+					cancelPlacements = append(cancelPlacements, &dextypes.Cancellation{
+						Id:                order.Id,
+						Initiator:         dextypes.CancellationInitiator_USER,
+						Creator:           order.Account,
+						ContractAddr:      order.ContractAddr,
+						PriceDenom:        order.PriceDenom,
+						AssetDenom:        order.AssetDenom,
+						PositionDirection: order.PositionDirection,
+						Price:             order.Price,
+					})
+				}
+			}
+
+			msg = &dextypes.MsgCancelOrders{
+				Creator:       sdk.AccAddress(key.PubKey().Address()).String(),
+				Cancellations: cancelPlacements,
+				ContractAddr:  contract,
+			}
+		default:
+			panic("Unknown message type")
 		}
+
 	default:
-		fmt.Printf("Unrecognized message type %s", config.MessageType)
+		fmt.Printf("Unrecognized message type %s", c.LoadTestConfig.MessageType)
 	}
 	return msg
 }