From e946bfb6969d8425e6baf556532a28a6d040d8f1 Mon Sep 17 00:00:00 2001 From: Steven Landers Date: Wed, 20 Mar 2024 15:05:23 -0400 Subject: [PATCH] [OCC] if no txs, avoid scheduler overhead, limit tasks (#468) ## Describe your changes and provide context - avoids scheduler if no txs (mainly for gassless situations) - limits tasks to the min(task count, workers) ## Testing performed to validate your change - unit tests and load testing --- baseapp/abci.go | 19 +++++++++------- baseapp/deliver_tx_batch_test.go | 38 ++++++++++++++++++++++++++++++++ tasks/scheduler.go | 2 +- 3 files changed, 50 insertions(+), 9 deletions(-) diff --git a/baseapp/abci.go b/baseapp/abci.go index 7e528d121..c005111ea 100644 --- a/baseapp/abci.go +++ b/baseapp/abci.go @@ -12,8 +12,6 @@ import ( "syscall" "time" - "github.com/cosmos/cosmos-sdk/tasks" - "github.com/armon/go-metrics" "github.com/gogo/protobuf/proto" abci "github.com/tendermint/tendermint/abci/types" @@ -23,6 +21,7 @@ import ( "github.com/cosmos/cosmos-sdk/codec" snapshottypes "github.com/cosmos/cosmos-sdk/snapshots/types" + "github.com/cosmos/cosmos-sdk/tasks" "github.com/cosmos/cosmos-sdk/telemetry" sdk "github.com/cosmos/cosmos-sdk/types" sdkerrors "github.com/cosmos/cosmos-sdk/types/errors" @@ -252,19 +251,23 @@ func (app *BaseApp) CheckTx(ctx context.Context, req *abci.RequestCheckTx) (*abc // DeliverTxBatch executes multiple txs func (app *BaseApp) DeliverTxBatch(ctx sdk.Context, req sdk.DeliverTxBatchRequest) (res sdk.DeliverTxBatchResponse) { - scheduler := tasks.NewScheduler(app.concurrencyWorkers, app.TracingInfo, app.DeliverTx) - // This will basically no-op the actual prefill if the metadata for the txs is empty + responses := make([]*sdk.DeliverTxResult, 0, len(req.TxEntries)) + + if len(req.TxEntries) == 0 { + return sdk.DeliverTxBatchResponse{Results: responses} + } - // process all txs, this will also initializes the MVS if prefill estimates was disabled + // avoid overhead for empty batches + scheduler := tasks.NewScheduler(app.concurrencyWorkers, app.TracingInfo, app.DeliverTx) txRes, err := scheduler.ProcessAll(ctx, req.TxEntries) if err != nil { - // TODO: handle error + ctx.Logger().Error("error while processing scheduler", "err", err) + panic(err) } - - responses := make([]*sdk.DeliverTxResult, 0, len(req.TxEntries)) for _, tx := range txRes { responses = append(responses, &sdk.DeliverTxResult{Response: tx}) } + return sdk.DeliverTxBatchResponse{Results: responses} } diff --git a/baseapp/deliver_tx_batch_test.go b/baseapp/deliver_tx_batch_test.go index 3cf6e0739..16ece7769 100644 --- a/baseapp/deliver_tx_batch_test.go +++ b/baseapp/deliver_tx_batch_test.go @@ -144,3 +144,41 @@ func TestDeliverTxBatch(t *testing.T) { app.Commit(context.Background()) } } + +func TestDeliverTxBatchEmpty(t *testing.T) { + // test increments in the ante + anteKey := []byte("ante-key") + + anteOpt := func(bapp *BaseApp) { + bapp.SetAnteHandler(anteHandler(capKey1, anteKey)) + } + + // test increments in the handler + routerOpt := func(bapp *BaseApp) { + r := sdk.NewRoute(routeMsgCounter, handlerKVStore(capKey1)) + bapp.Router().AddRoute(r) + } + + app := setupBaseApp(t, anteOpt, routerOpt) + app.InitChain(context.Background(), &abci.RequestInitChain{}) + + // Create same codec used in txDecoder + codec := codec.NewLegacyAmino() + registerTestCodec(codec) + + nBlocks := 3 + for blockN := 0; blockN < nBlocks; blockN++ { + header := tmproto.Header{Height: int64(blockN) + 1} + app.setDeliverState(header) + app.BeginBlock(app.deliverState.ctx, abci.RequestBeginBlock{Header: header}) + + var requests []*sdk.DeliverTxEntry + responses := app.DeliverTxBatch(app.deliverState.ctx, sdk.DeliverTxBatchRequest{TxEntries: requests}) + require.Len(t, responses.Results, 0) + + app.EndBlock(app.deliverState.ctx, abci.RequestEndBlock{}) + require.Empty(t, app.deliverState.ctx.MultiStore().GetEvents()) + app.SetDeliverStateToCommit() + app.Commit(context.Background()) + } +} diff --git a/tasks/scheduler.go b/tasks/scheduler.go index 90d499927..4b2f45a68 100644 --- a/tasks/scheduler.go +++ b/tasks/scheduler.go @@ -286,7 +286,7 @@ func (s *scheduler) ProcessAll(ctx sdk.Context, reqs []*sdk.DeliverTxEntry) ([]t // default to number of tasks if workers is negative or 0 by this point workers := s.workers - if s.workers < 1 { + if s.workers < 1 || len(tasks) < s.workers { workers = len(tasks) }