Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Parallelize Sudo Deposit #777

Merged
merged 6 commits into from
May 15, 2023
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 50 additions & 8 deletions x/dex/contract/abci.go
Original file line number Diff line number Diff line change
Expand Up @@ -161,22 +161,64 @@ func decorateContextForContract(ctx sdk.Context, contractInfo types.ContractInfo
)
}

func getSudoPlaceOrderMessages(sdkCtx sdk.Context, validContractsInfos []types.ContractInfoV2, keeperWrapper dexkeeperabci.KeeperWrapper) map[int]types.SudoOrderPlacementMsg {
resultChan := make(chan struct {
int
types.SudoOrderPlacementMsg
}, len(validContractsInfos))

var wg sync.WaitGroup
contractToProcess := dexutils.GetMemState(sdkCtx.Context()).GetContractToProcess()
for index, contract := range validContractsInfos {
wg.Add(1)
go func(index int, contract types.ContractInfoV2) {
defer wg.Done()
if !contract.NeedOrderMatching || contractToProcess.Contains(contract.ContractAddr) {
return
}
typedContractAddr := types.ContractAddress(contract.ContractAddr)
msg := keeperWrapper.GetDepositSudoMsg(sdkCtx, typedContractAddr)
if msg.IsEmpty() {
return
}
resultChan <- struct {
int
types.SudoOrderPlacementMsg
}{index, msg}

}(index, contract)
Fixed Show fixed Hide fixed
}

go func() {
wg.Wait()
close(resultChan)
}()
Comment on lines +192 to +195

Check notice

Code scanning / CodeQL

Spawning a Go routine

Spawning a Go routine may be a possible source of non-determinism

resultMapToIndex := map[int]types.SudoOrderPlacementMsg{}
for result := range resultChan {
resultMapToIndex[result.int] = result.SudoOrderPlacementMsg
}

keeperWrapper.BankKeeper.WriteDeferredOperations(sdkCtx)

return resultMapToIndex
}

func handleDeposits(spanCtx context.Context, ctx sdk.Context, env *environment, keeper *keeper.Keeper, tracer *otrace.Tracer) {
// Handle deposit sequentially since they mutate `bank` state which is shared by all contracts
_, span := (*tracer).Start(spanCtx, "handleDeposits")
defer span.End()
defer telemetry.MeasureSince(time.Now(), "dex", "handle_deposits")
keeperWrapper := dexkeeperabci.KeeperWrapper{Keeper: keeper}
for _, contract := range env.validContractsInfo {
if !dexutils.GetMemState(ctx.Context()).GetContractToProcess().Contains(contract.ContractAddr) {
continue
}
if !contract.NeedOrderMatching {
continue
}
if err := keeperWrapper.HandleEBDeposit(spanCtx, ctx, tracer, contract.ContractAddr); err != nil {

// Filter and create and pending sudo messages concurrently and batch write to bank keeper
sudoMessages := getSudoPlaceOrderMessages(ctx, env.validContractsInfo, keeperWrapper)
for index, message := range sudoMessages {
contract := env.validContractsInfo[index]
if err := keeperWrapper.HandleEBDeposit(spanCtx, ctx, tracer, contract.ContractAddr, message); err != nil {
env.addError(contract.ContractAddr, err)
}

}
}

Expand Down
22 changes: 14 additions & 8 deletions x/dex/keeper/abci/end_block_deposit.go
Original file line number Diff line number Diff line change
Expand Up @@ -13,17 +13,12 @@ import (
otrace "go.opentelemetry.io/otel/trace"
)

func (w KeeperWrapper) HandleEBDeposit(ctx context.Context, sdkCtx sdk.Context, tracer *otrace.Tracer, contractAddr string) error {
func (w KeeperWrapper) HandleEBDeposit(ctx context.Context, sdkCtx sdk.Context, tracer *otrace.Tracer, contractAddr string, message types.SudoOrderPlacementMsg) error {
_, span := (*tracer).Start(ctx, "SudoDeposit")
span.SetAttributes(attribute.String("contractAddr", contractAddr))
defer span.End()

typedContractAddr := types.ContractAddress(contractAddr)
msg := w.GetDepositSudoMsg(sdkCtx, typedContractAddr)
if msg.IsEmpty() {
return nil
}
_, err := utils.CallContractSudo(sdkCtx, w.Keeper, contractAddr, msg, dexutils.ZeroUserProvidedGas) // deposit
_, err := utils.CallContractSudo(sdkCtx, w.Keeper, contractAddr, message, dexutils.ZeroUserProvidedGas) // deposit
if err != nil {
sdkCtx.Logger().Error(fmt.Sprintf("Error during deposit: %s", err.Error()))
return err
Expand All @@ -34,6 +29,17 @@ func (w KeeperWrapper) HandleEBDeposit(ctx context.Context, sdkCtx sdk.Context,

func (w KeeperWrapper) GetDepositSudoMsg(ctx sdk.Context, typedContractAddr types.ContractAddress) types.SudoOrderPlacementMsg {
depositMemState := dexutils.GetMemState(ctx.Context()).GetDepositInfo(ctx, typedContractAddr).Get()

// If there's no amount to send, exit early and avoid additional processing
if len(depositMemState) == 0 {
return types.SudoOrderPlacementMsg{
OrderPlacements: types.OrderPlacementMsgDetails{
Orders: []types.Order{},
Deposits: []types.ContractDepositInfo{},
},
}
}

contractDepositInfo := seiutils.Map(
depositMemState,
func(d *types.DepositInfoEntry) types.ContractDepositInfo { return d.ToContractDepositInfo() },
Expand All @@ -46,7 +52,7 @@ func (w KeeperWrapper) GetDepositSudoMsg(ctx sdk.Context, typedContractAddr type
if err != nil {
panic(err)
}
if err := w.BankKeeper.SendCoinsFromModuleToAccount(ctx, types.ModuleName, contractAddr, escrowed); err != nil {
if err := w.BankKeeper.DeferredSendCoinsFromModuleToAccount(ctx, types.ModuleName, contractAddr, escrowed); err != nil {
panic(err)
}
return types.SudoOrderPlacementMsg{
Expand Down
4 changes: 2 additions & 2 deletions x/oracle/simulation/operations.go
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,7 @@ func WeightedOperations(
}

// 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,
Expand Down Expand Up @@ -122,7 +122,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,
Expand Down