matching: refactor taskListManager to separate out individual compone…

…nts (cadence-workflow#2125)

common/util.go

@@ -322,6 +322,14 @@ func MinInt64(a, b int64) int64 {
 	return b
 }
 
+// MaxInt64 returns the greater of two given int64
+func MaxInt64(a, b int64) int64 {
+	if a > b {
+		return a
+	}
+	return b
+}
+
 // MinInt32 return smaller one of two inputs int32
 func MinInt32(a, b int32) int32 {
 	if a < b {

service/matching/config.go

@@ -0,0 +1,134 @@
+// Copyright (c) 2019 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package matching
+
+import (
+	"time"
+
+	"github.com/uber/cadence/common/cache"
+	"github.com/uber/cadence/common/service/dynamicconfig"
+)
+
+type (
+	// Config represents configuration for cadence-matching service
+	Config struct {
+		PersistenceMaxQPS dynamicconfig.IntPropertyFn
+		EnableSyncMatch   dynamicconfig.BoolPropertyFnWithTaskListInfoFilters
+		RPS               dynamicconfig.IntPropertyFn
+
+		// taskListManager configuration
+		RangeSize                 int64
+		GetTasksBatchSize         dynamicconfig.IntPropertyFnWithTaskListInfoFilters
+		UpdateAckInterval         dynamicconfig.DurationPropertyFnWithTaskListInfoFilters
+		IdleTasklistCheckInterval dynamicconfig.DurationPropertyFnWithTaskListInfoFilters
+		MaxTasklistIdleTime       dynamicconfig.DurationPropertyFnWithTaskListInfoFilters
+		// Time to hold a poll request before returning an empty response if there are no tasks
+		LongPollExpirationInterval dynamicconfig.DurationPropertyFnWithTaskListInfoFilters
+		MinTaskThrottlingBurstSize dynamicconfig.IntPropertyFnWithTaskListInfoFilters
+		MaxTaskDeleteBatchSize     dynamicconfig.IntPropertyFnWithTaskListInfoFilters
+
+		// taskWriter configuration
+		OutstandingTaskAppendsThreshold dynamicconfig.IntPropertyFnWithTaskListInfoFilters
+		MaxTaskBatchSize                dynamicconfig.IntPropertyFnWithTaskListInfoFilters
+
+		ThrottledLogRPS dynamicconfig.IntPropertyFn
+	}
+
+	taskListConfig struct {
+		EnableSyncMatch func() bool
+		// Time to hold a poll request before returning an empty response if there are no tasks
+		LongPollExpirationInterval func() time.Duration
+		RangeSize                  int64
+		GetTasksBatchSize          func() int
+		UpdateAckInterval          func() time.Duration
+		IdleTasklistCheckInterval  func() time.Duration
+		MaxTasklistIdleTime        func() time.Duration
+		MinTaskThrottlingBurstSize func() int
+		MaxTaskDeleteBatchSize     func() int
+		// taskWriter configuration
+		OutstandingTaskAppendsThreshold func() int
+		MaxTaskBatchSize                func() int
+	}
+)
+
+// NewConfig returns new service config with default values
+func NewConfig(dc *dynamicconfig.Collection) *Config {
+	return &Config{
+		PersistenceMaxQPS:               dc.GetIntProperty(dynamicconfig.MatchingPersistenceMaxQPS, 3000),
+		EnableSyncMatch:                 dc.GetBoolPropertyFilteredByTaskListInfo(dynamicconfig.MatchingEnableSyncMatch, true),
+		RPS:                             dc.GetIntProperty(dynamicconfig.MatchingRPS, 1200),
+		RangeSize:                       100000,
+		GetTasksBatchSize:               dc.GetIntPropertyFilteredByTaskListInfo(dynamicconfig.MatchingGetTasksBatchSize, 1000),
+		UpdateAckInterval:               dc.GetDurationPropertyFilteredByTaskListInfo(dynamicconfig.MatchingUpdateAckInterval, 1*time.Minute),
+		IdleTasklistCheckInterval:       dc.GetDurationPropertyFilteredByTaskListInfo(dynamicconfig.MatchingIdleTasklistCheckInterval, 5*time.Minute),
+		MaxTasklistIdleTime:             dc.GetDurationPropertyFilteredByTaskListInfo(dynamicconfig.MaxTasklistIdleTime, 5*time.Minute),
+		LongPollExpirationInterval:      dc.GetDurationPropertyFilteredByTaskListInfo(dynamicconfig.MatchingLongPollExpirationInterval, time.Minute),
+		MinTaskThrottlingBurstSize:      dc.GetIntPropertyFilteredByTaskListInfo(dynamicconfig.MatchingMinTaskThrottlingBurstSize, 1),
+		MaxTaskDeleteBatchSize:          dc.GetIntPropertyFilteredByTaskListInfo(dynamicconfig.MatchingMaxTaskDeleteBatchSize, 100),
+		OutstandingTaskAppendsThreshold: dc.GetIntPropertyFilteredByTaskListInfo(dynamicconfig.MatchingOutstandingTaskAppendsThreshold, 250),
+		MaxTaskBatchSize:                dc.GetIntPropertyFilteredByTaskListInfo(dynamicconfig.MatchingMaxTaskBatchSize, 100),
+		ThrottledLogRPS:                 dc.GetIntProperty(dynamicconfig.MatchingThrottledLogRPS, 20),
+	}
+}
+
+func newTaskListConfig(id *taskListID, config *Config, domainCache cache.DomainCache) (*taskListConfig, error) {
+	domainEntry, err := domainCache.GetDomainByID(id.domainID)
+	if err != nil {
+		return nil, err
+	}
+
+	domain := domainEntry.GetInfo().Name
+	taskListName := id.taskListName
+	taskType := id.taskType
+	return &taskListConfig{
+		RangeSize: config.RangeSize,
+		GetTasksBatchSize: func() int {
+			return config.GetTasksBatchSize(domain, taskListName, taskType)
+		},
+		UpdateAckInterval: func() time.Duration {
+			return config.UpdateAckInterval(domain, taskListName, taskType)
+		},
+		IdleTasklistCheckInterval: func() time.Duration {
+			return config.IdleTasklistCheckInterval(domain, taskListName, taskType)
+		},
+		MaxTasklistIdleTime: func() time.Duration {
+			return config.MaxTasklistIdleTime(domain, taskListName, taskType)
+		},
+		MinTaskThrottlingBurstSize: func() int {
+			return config.MinTaskThrottlingBurstSize(domain, taskListName, taskType)
+		},
+		EnableSyncMatch: func() bool {
+			return config.EnableSyncMatch(domain, taskListName, taskType)
+		},
+		LongPollExpirationInterval: func() time.Duration {
+			return config.LongPollExpirationInterval(domain, taskListName, taskType)
+		},
+		MaxTaskDeleteBatchSize: func() int {
+			return config.MaxTaskDeleteBatchSize(domain, taskListName, taskType)
+		},
+		OutstandingTaskAppendsThreshold: func() int {
+			return config.OutstandingTaskAppendsThreshold(domain, taskListName, taskType)
+		},
+		MaxTaskBatchSize: func() int {
+			return config.MaxTaskBatchSize(domain, taskListName, taskType)
+		},
+	}, nil
+}

service/matching/handler.go

@@ -129,7 +129,7 @@ func (h *Handler) AddActivityTask(ctx context.Context, addRequest *m.AddActivity
 		return h.handleErr(errMatchingHostThrottle, scope)
 	}
 
-	syncMatch, err := h.engine.AddActivityTask(addRequest)
+	syncMatch, err := h.engine.AddActivityTask(ctx, addRequest)
 	if syncMatch {
 		h.metricsClient.RecordTimer(scope, metrics.SyncMatchLatency, time.Since(startT))
 	}
@@ -149,7 +149,7 @@ func (h *Handler) AddDecisionTask(ctx context.Context, addRequest *m.AddDecision
 		return h.handleErr(errMatchingHostThrottle, scope)
 	}
 
-	syncMatch, err := h.engine.AddDecisionTask(addRequest)
+	syncMatch, err := h.engine.AddDecisionTask(ctx, addRequest)
 	if syncMatch {
 		h.metricsClient.RecordTimer(scope, metrics.SyncMatchLatency, time.Since(startT))
 	}

service/matching/matcher.go

@@ -0,0 +1,203 @@
+// Copyright (c) 2019 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package matching
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/uber/cadence/common/metrics"
+	"golang.org/x/time/rate"
+)
+
+// TaskMatcher matches a task producer with a task consumer
+// Producers are usually rpc calls from history or taskReader
+// that drains backlog from db. Consumers are the task list pollers
+type TaskMatcher struct {
+	// synchronous task channel to match producer/consumer
+	taskC chan *internalTask
+	// synchronous task channel to match query task - the reason to have
+	// separate channel for this is because there are cases when consumers
+	// are interested in queryTasks but not others. Example is when domain is
+	// not active in a cluster
+	queryTaskC chan *internalTask
+	// ratelimiter that limits the rate at which tasks can be dispatched to consumers
+	limiter *rateLimiter
+	// domain metric scope
+	scope func() metrics.Scope
+}
+
+const (
+	_defaultTaskDispatchRPS    = 100000.0
+	_defaultTaskDispatchRPSTTL = 60 * time.Second
+)
+
+var errTasklistThrottled = errors.New("cannot add to tasklist, limit exceeded")
+
+// newTaskMatcher returns an task matcher instance. The returned instance can be
+// used by task producers and consumers to find a match. Both sync matches and non-sync
+// matches should use this implementation
+func newTaskMatcher(config *taskListConfig, scopeFunc func() metrics.Scope) *TaskMatcher {
+	dPtr := _defaultTaskDispatchRPS
+	limiter := newRateLimiter(&dPtr, _defaultTaskDispatchRPSTTL, config.MinTaskThrottlingBurstSize())
+	return &TaskMatcher{
+		limiter:    limiter,
+		scope:      scopeFunc,
+		taskC:      make(chan *internalTask),
+		queryTaskC: make(chan *internalTask),
+	}
+}
+
+// Offer offers a task to a potential consumer (poller)
+// If the task is successfully matched with a consumer, this
+// method will return true and no error. If the task is matched
+// but consumer returned error, then this method will return
+// true and error message. Both regular tasks and query tasks
+// should use this method to match with a consumer. Likewise, sync matches
+// and non-sync matches both should use this method.
+// returns error when:
+//  - ratelimit is exceeded (does not apply to query task)
+//  - context deadline is exceeded
+//  - task is matched and consumer returns error in response channel
+func (tm *TaskMatcher) Offer(ctx context.Context, task *internalTask) (bool, error) {
+	if task.isQuery() {
+		select {
+		case tm.queryTaskC <- task:
+			<-task.syncResponseCh
+			return true, nil
+		case <-ctx.Done():
+			return false, ctx.Err()
+		}
+	}
+
+	rsv, err := tm.ratelimit(ctx)
+	if err != nil {
+		tm.scope().IncCounter(metrics.SyncThrottleCounter)
+		return false, err
+	}
+
+	select {
+	case tm.taskC <- task: // poller picked up the task
+		if task.syncResponseCh != nil {
+			// if there is a response channel, block until resp is received
+			// and return error if the response contains error
+			err = <-task.syncResponseCh
+			return true, err
+		}
+		return false, nil
+	default: // no poller waiting for tasks
+		if rsv != nil {
+			// there was a ratelimit token we consumed
+			// return it since we did not really do any work
+			rsv.Cancel()
+		}
+		return false, nil
+	}
+}
+
+// MustOffer blocks until a consumer is found to handle this task
+// Returns error only when context is canceled or the ratelimit is set to zero (allow nothing)
+// The passed in context MUST NOT have a deadline associated with it
+func (tm *TaskMatcher) MustOffer(ctx context.Context, task *internalTask) error {
+	if _, err := tm.ratelimit(ctx); err != nil {
+		return err
+	}
+	select {
+	case tm.taskC <- task:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
+// Poll blocks until a task is found or context deadline is exceeded
+// On success, the returned task could be a query task or a regular task
+// Returns ErrNoTasks when context deadline is exceeded
+func (tm *TaskMatcher) Poll(ctx context.Context) (*internalTask, error) {
+	select {
+	case task := <-tm.taskC:
+		if task.syncResponseCh != nil {
+			tm.scope().IncCounter(metrics.PollSuccessWithSyncCounter)
+		}
+		tm.scope().IncCounter(metrics.PollSuccessCounter)
+		return task, nil
+	case task := <-tm.queryTaskC:
+		tm.scope().IncCounter(metrics.PollSuccessWithSyncCounter)
+		tm.scope().IncCounter(metrics.PollSuccessCounter)
+		return task, nil
+	case <-ctx.Done():
+		tm.scope().IncCounter(metrics.PollTimeoutCounter)
+		return nil, ErrNoTasks
+	}
+}
+
+// PollForQuery blocks until a *query* task is found or context deadline is exceeded
+// Returns ErrNoTasks when context deadline is exceeded
+func (tm *TaskMatcher) PollForQuery(ctx context.Context) (*internalTask, error) {
+	select {
+	case task := <-tm.queryTaskC:
+		tm.scope().IncCounter(metrics.PollSuccessWithSyncCounter)
+		tm.scope().IncCounter(metrics.PollSuccessCounter)
+		return task, nil
+	case <-ctx.Done():
+		tm.scope().IncCounter(metrics.PollTimeoutCounter)
+		return nil, ErrNoTasks
+	}
+}
+
+// UpdateRatelimit updates the task dispatch rate
+func (tm *TaskMatcher) UpdateRatelimit(rps *float64) {
+	tm.limiter.UpdateMaxDispatch(rps)
+}
+
+// Rate returns the current rate at which tasks are dispatched
+func (tm *TaskMatcher) Rate() float64 {
+	return tm.limiter.Limit()
+}
+
+func (tm *TaskMatcher) ratelimit(ctx context.Context) (*rate.Reservation, error) {
+	select {
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	default:
+	}
+
+	deadline, ok := ctx.Deadline()
+	if !ok {
+		if err := tm.limiter.Wait(ctx); err != nil {
+			return nil, err
+		}
+		return nil, nil
+	}
+
+	rsv := tm.limiter.Reserve()
+	// If we have to wait too long for reservation, give up and return
+	if !rsv.OK() || rsv.Delay() > deadline.Sub(time.Now()) {
+		if rsv.OK() { // if we were indeed given a reservation, return it before we bail out
+			rsv.Cancel()
+		}
+		return nil, errTasklistThrottled
+	}
+
+	time.Sleep(rsv.Delay())
+	return rsv, nil
+}