[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

@@ -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}
 }
 

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())
+	}
+}

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)
 	}