Expose worker option to set ConcurrentDecisionTaskExecutionSize (#539)

Author: yiminc

internal/internal_worker.go

@@ -61,7 +61,7 @@ const (
 	defaultTaskListActivitiesPerSecond = 100000.0 // Large activity executions/sec (unlimited)
 
 	defaultMaxConcurrentTaskExecutionSize = 1000   // hardcoded max task execution size.
-	defaultMaxTaskExecutionRate           = 100000 // Large task execution rate (unlimited)
+	defaultWorkerTaskExecutionRate        = 100000 // Large task execution rate (unlimited)
 
 	defaultPollerRate = 1000
 
@@ -116,6 +116,12 @@ type (
 		// Defines rate limiting on number of activity tasks that can be executed per second per worker.
 		WorkerActivitiesPerSecond float64
 
+		// Defines how many concurrent decision task executions by this worker.
+		ConcurrentDecisionTaskExecutionSize int
+
+		// Defines rate limiting on number of decision tasks that can be executed per second per worker.
+		WorkerDecisionTasksPerSecond float64
+
 		// Defines how many concurrent local activity executions by this worker.
 		ConcurrentLocalActivityExecutionSize int
 
@@ -257,8 +263,8 @@ func newWorkflowTaskWorkerInternal(
 	worker := newBaseWorker(baseWorkerOptions{
 		pollerCount:       params.ConcurrentPollRoutineSize,
 		pollerRate:        defaultPollerRate,
-		maxConcurrentTask: defaultMaxConcurrentTaskExecutionSize,
-		maxTaskPerSecond:  defaultMaxTaskExecutionRate,
+		maxConcurrentTask: params.ConcurrentDecisionTaskExecutionSize,
+		maxTaskPerSecond:  params.WorkerDecisionTasksPerSecond,
 		taskWorker:        poller,
 		identity:          params.Identity,
 		workerType:        "DecisionWorker"},
@@ -931,6 +937,8 @@ func newAggregatedWorker(
 		WorkerActivitiesPerSecond:            wOptions.WorkerActivitiesPerSecond,
 		ConcurrentLocalActivityExecutionSize: wOptions.MaxConcurrentLocalActivityExecutionSize,
 		WorkerLocalActivitiesPerSecond:       wOptions.WorkerLocalActivitiesPerSecond,
+		ConcurrentDecisionTaskExecutionSize:  wOptions.MaxConcurrentDecisionTaskExecutionSize,
+		WorkerDecisionTasksPerSecond:         wOptions.WorkerDecisionTasksPerSecond,
 		Identity:                             wOptions.Identity,
 		MetricsScope:                         wOptions.MetricsScope,
 		Logger:                               wOptions.Logger,
@@ -1151,6 +1159,12 @@ func fillWorkerOptionsDefaults(options WorkerOptions) WorkerOptions {
 	if options.WorkerActivitiesPerSecond == 0 {
 		options.WorkerActivitiesPerSecond = defaultWorkerActivitiesPerSecond
 	}
+	if options.MaxConcurrentDecisionTaskExecutionSize == 0 {
+		options.MaxConcurrentDecisionTaskExecutionSize = defaultMaxConcurrentTaskExecutionSize
+	}
+	if options.WorkerDecisionTasksPerSecond == 0 {
+		options.WorkerDecisionTasksPerSecond = defaultWorkerTaskExecutionRate
+	}
 	if options.MaxConcurrentLocalActivityExecutionSize == 0 {
 		options.MaxConcurrentLocalActivityExecutionSize = defaultMaxConcurrentLocalActivityExecutionSize
 	}

internal/worker.go

@@ -90,6 +90,16 @@ type (
 		// The zero value of this uses the default value. Default: 100k
 		TaskListActivitiesPerSecond float64
 
+		// Optional: To set the maximum concurrent decision task executions this worker can have.
+		// The zero value of this uses the default value.
+		// default: defaultMaxConcurrentTaskExecutionSize(1k)
+		MaxConcurrentDecisionTaskExecutionSize int
+
+		// Optional: Sets the rate limiting on number of decision tasks that can be executed per second per
+		// worker. This can be used to limit resources used by the worker.
+		// The zero value of this uses the default value. Default: 100k
+		WorkerDecisionTasksPerSecond float64
+
 		// Optional: if the activities need auto heart beating for those activities
 		// by the framework
 		// default: false not to heartbeat.