Restrict the number of results processed per index worker

src/cmd/services/m3dbnode/config/config.go

@@ -387,6 +387,13 @@ type IndexConfiguration struct {
 	// as they are very CPU-intensive (regex and FST matching).
 	MaxQueryIDsConcurrency int `yaml:"maxQueryIDsConcurrency" validate:"min=0"`
 
+	// MaxWorkerTime is the maximum time a query can hold an index worker at once. If a query does not finish in this
+	// time it yields the worker and must wait again for another worker to resume. The number of workers available to
+	// all queries is defined by MaxQueryIDsConcurrency.
+	// Capping the maximum time per worker ensures a few large queries don't hold all the concurrent workers and lock
+	// out many small queries from running.
+	MaxWorkerTime time.Duration `yaml:"maxWorkerTime"`
+
 	// RegexpDFALimit is the limit on the max number of states used by a
 	// regexp deterministic finite automaton. Default is 10,000 states.
 	RegexpDFALimit *int `yaml:"regexpDFALimit"`
@@ -412,6 +419,14 @@ type IndexConfiguration struct {
 	ForwardIndexThreshold float64 `yaml:"forwardIndexThreshold" validate:"min=0.0,max=1.0"`
 }
 
+// MaxResultsPerWorkerConfiguration configures the max results per index worker.
+type MaxResultsPerWorkerConfiguration struct {
+	// Fetch is the max for fetch queries.
+	Fetch int `yaml:"fetch"`
+	// Aggregate is the max for aggregate queries.
+	Aggregate int `yaml:"aggregate"`
+}
+
 // RegexpDFALimitOrDefault returns the deterministic finite automaton states
 // limit or default.
 func (c IndexConfiguration) RegexpDFALimitOrDefault() int {

src/cmd/services/m3dbnode/config/config_test.go

@@ -339,6 +339,7 @@ func TestConfiguration(t *testing.T) {
 	expected := `db:
   index:
     maxQueryIDsConcurrency: 0
+    maxWorkerTime: 0s
     regexpDFALimit: null
     regexpFSALimit: null
     forwardIndexProbability: 0

src/dbnode/network/server/tchannelthrift/node/fetch_result_iter_test.go

@@ -201,7 +201,7 @@ func (p *fakePermits) TryAcquire(_ context.Context) (bool, error) {
 	return true, nil
 }
 
-func (p *fakePermits) Release() {
+func (p *fakePermits) Release(_ int64) {
 	p.released++
 	p.available++
 }

src/dbnode/network/server/tchannelthrift/node/service.go

@@ -1092,9 +1092,10 @@ func (i *fetchTaggedResultsIter) Close(err error) {
 
 	i.seriesBlocks.RecordValue(float64(i.totalSeriesBlocks))
 
-	for n := 0; n < i.batchesAcquired; n++ {
-		i.blockPermits.Release()
+	for n := 0; n < i.batchesAcquired-1; n++ {
+		i.blockPermits.Release(int64(i.blocksPerBatch))
 	}
+	i.blockPermits.Release(int64(i.blocksPerBatch - i.blocksAvailable))
 }
 
 // IDResult is the FetchTagged result for a series ID.

src/dbnode/server/server.go

@@ -90,13 +90,12 @@ import (
 	xos "github.com/m3db/m3/src/x/os"
 	"github.com/m3db/m3/src/x/pool"
 	"github.com/m3db/m3/src/x/serialize"
-	xsync "github.com/m3db/m3/src/x/sync"
 
 	apachethrift "github.com/apache/thrift/lib/go/thrift"
 	"github.com/m3dbx/vellum/levenshtein"
 	"github.com/m3dbx/vellum/levenshtein2"
 	"github.com/m3dbx/vellum/regexp"
-	opentracing "github.com/opentracing/opentracing-go"
+	"github.com/opentracing/opentracing-go"
 	"github.com/uber-go/tally"
 	"github.com/uber/tchannel-go"
 	"go.etcd.io/etcd/embed"
@@ -371,14 +370,6 @@ func Run(runOpts RunOptions) {
 
 	opentracing.SetGlobalTracer(tracer)
 
-	if cfg.Index.MaxQueryIDsConcurrency != 0 {
-		queryIDsWorkerPool := xsync.NewWorkerPool(cfg.Index.MaxQueryIDsConcurrency)
-		queryIDsWorkerPool.Init()
-		opts = opts.SetQueryIDsWorkerPool(queryIDsWorkerPool)
-	} else {
-		logger.Warn("max index query IDs concurrency was not set, falling back to default value")
-	}
-
 	// Set global index options.
 	if n := cfg.Index.RegexpDFALimitOrDefault(); n > 0 {
 		regexp.SetStateLimit(n)
@@ -481,8 +472,14 @@ func Run(runOpts RunOptions) {
 	seriesReadPermits.Start()
 	defer seriesReadPermits.Stop()
 
-	opts = opts.SetPermitsOptions(opts.PermitsOptions().
-		SetSeriesReadPermitsManager(seriesReadPermits))
+	permitOptions := opts.PermitsOptions().SetSeriesReadPermitsManager(seriesReadPermits)
+	if cfg.Index.MaxQueryIDsConcurrency != 0 {
+		permitOptions = permitOptions.SetIndexQueryPermitsManager(
+			permits.NewFixedPermitsManager(cfg.Index.MaxQueryIDsConcurrency))
+	} else {
+		logger.Warn("max index query IDs concurrency was not set, falling back to default value")
+	}
+	opts = opts.SetPermitsOptions(permitOptions)
 
 	// Setup postings list cache.
 	var (
@@ -524,6 +521,11 @@ func Run(runOpts RunOptions) {
 		}).
 		SetMmapReporter(mmapReporter).
 		SetQueryLimits(queryLimits)
+
+	if cfg.Index.MaxWorkerTime > 0 {
+		indexOpts = indexOpts.SetMaxWorkerTime(cfg.Index.MaxWorkerTime)
+	}
+
 	opts = opts.SetIndexOptions(indexOpts)
 
 	if tick := cfg.Tick; tick != nil {