Exclude no work executor

For executor which has nothing to do in a given segment, we should not
include it in the list of executors.

To do it, add new method HasWork to ExecutorConfig. By filtering out no
work executor when creating new schedulter, we can prevents any un-necessary
works and provide better UX.

Fixes #1295

cmd/run.go

@@ -228,17 +228,17 @@ a commandline interface for interacting with it.`,
 			fprintf(stdout, "\n")
 
 			plan := execScheduler.GetExecutionPlan()
-			executors := execScheduler.GetExecutors()
+			executorConfigs := execScheduler.GetExecutorConfigs()
 			maxDuration, _ := lib.GetEndOffset(plan)
 
 			fprintf(stdout, "  execution: %s\n", ui.ValueColor.Sprintf(
-				"(%.2f%%) %d executors, %d max VUs, %s max duration (incl. graceful stop):",
-				conf.ExecutionSegment.FloatLength()*100, len(executors),
+				"(%.2f%%) %d executorConfigs, %d max VUs, %s max duration (incl. graceful stop):",
+				conf.ExecutionSegment.FloatLength()*100, len(executorConfigs),
 				lib.GetMaxPossibleVUs(plan), maxDuration),
 			)
-			for _, sched := range executors {
+			for _, ec := range executorConfigs {
 				fprintf(stdout, "           * %s: %s\n",
-					sched.GetConfig().GetName(), sched.GetConfig().GetDescription(conf.ExecutionSegment))
+					ec.GetName(), ec.GetDescription(conf.ExecutionSegment))
 			}
 			fprintf(stdout, "\n")
 		}

core/local/local.go

@@ -40,12 +40,13 @@ type ExecutionScheduler struct {
 	options lib.Options
 	logger  *logrus.Logger
 
-	initProgress   *pb.ProgressBar
-	executors      []lib.Executor // sorted by (startTime, ID)
-	executionPlan  []lib.ExecutionStep
-	maxDuration    time.Duration // cached value derived from the execution plan
-	maxPossibleVUs uint64        // cached value derived from the execution plan
-	state          *lib.ExecutionState
+	initProgress    *pb.ProgressBar
+	executors       []lib.Executor // sorted by (startTime, ID)
+	executorConfigs []lib.ExecutorConfig
+	executionPlan   []lib.ExecutionStep
+	maxDuration     time.Duration // cached value derived from the execution plan
+	maxPossibleVUs  uint64        // cached value derived from the execution plan
+	state           *lib.ExecutionState
 }
 
 // Check to see if we implement the lib.ExecutionScheduler interface
@@ -67,13 +68,20 @@ func NewExecutionScheduler(runner lib.Runner, logger *logrus.Logger) (*Execution
 
 	executorConfigs := options.Execution.GetSortedConfigs()
 	executors := make([]lib.Executor, len(executorConfigs))
-	for i, sc := range executorConfigs {
+	// Only take executor which has works.
+	n := 0
+	for _, sc := range executorConfigs {
+		if !sc.HasWork(executionState.Options.ExecutionSegment) {
+			continue
+		}
 		s, err := sc.NewExecutor(executionState, logger.WithField("executor", sc.GetName()))
 		if err != nil {
 			return nil, err
 		}
-		executors[i] = s
+		executors[n] = s
+		n++
 	}
+	executors = executors[:n]
 
 	if options.Paused.Bool {
 		if err := executionState.Pause(); err != nil {
@@ -86,12 +94,13 @@ func NewExecutionScheduler(runner lib.Runner, logger *logrus.Logger) (*Execution
 		logger:  logger,
 		options: options,
 
-		initProgress:   pb.New(pb.WithConstLeft("Init")),
-		executors:      executors,
-		executionPlan:  executionPlan,
-		maxDuration:    maxDuration,
-		maxPossibleVUs: maxPossibleVUs,
-		state:          executionState,
+		initProgress:    pb.New(pb.WithConstLeft("Init")),
+		executors:       executors,
+		executorConfigs: executorConfigs,
+		executionPlan:   executionPlan,
+		maxDuration:     maxDuration,
+		maxPossibleVUs:  maxPossibleVUs,
+		state:           executionState,
 	}, nil
 }
 
@@ -115,6 +124,11 @@ func (e *ExecutionScheduler) GetExecutors() []lib.Executor {
 	return e.executors
 }
 
+// GetExecutorConfigs returns the slice of executor configs.
+func (e *ExecutionScheduler) GetExecutorConfigs() []lib.ExecutorConfig {
+	return e.executorConfigs
+}
+
 // GetInitProgressBar returns the progress bar associated with the Init
 // function. After the Init is done, it is "hijacked" to display real-time
 // execution statistics as a text bar.

core/local/local_test.go

@@ -728,3 +728,56 @@ func TestSetPaused(t *testing.T) {
 		require.Contains(t, err.Error(), "doesn't support pause and resume operations after its start")
 	})
 }
+
+func TestNewExecutionSchedulerHasWork(t *testing.T) {
+	t.Parallel()
+	script := []byte(`
+import http from 'k6/http';
+
+export let options = {
+    executionSegment: "2/4:3/4",
+    execution: {
+        shared_iters1: {
+            type: "shared-iterations",
+            vus: 3,
+            iterations: 3,
+        },
+        shared_iters2: {
+            type: "shared-iterations",
+            vus: 4,
+            iterations: 4,
+        },
+        constant_arr_rate: {
+            type: "constant-arrival-rate",
+            rate: 3,
+            timeUnit: "1s",
+            duration: "20s",
+            preAllocatedVUs: 4,
+            maxVUs: 4,
+        },
+    },
+};
+
+export default function() {
+  const response = http.get("http://test.loadimpact.com");
+};`)
+
+	runner, err := js.New(
+		&loader.SourceData{
+			URL:  &url.URL{Path: "/script.js"},
+			Data: script,
+		},
+		nil,
+		lib.RuntimeOptions{},
+	)
+	require.NoError(t, err)
+
+	logger := logrus.New()
+	logger.SetOutput(testutils.NewTestOutput(t))
+
+	execScheduler, err := NewExecutionScheduler(runner, logger)
+	require.NoError(t, err)
+
+	assert.Len(t, execScheduler.executors, 2)
+	assert.Len(t, execScheduler.executorConfigs, 3)
+}

lib/executor/constant_arrival_rate.go

@@ -164,6 +164,11 @@ func (carc ConstantArrivalRateConfig) NewExecutor(
 	}, nil
 }
 
+// HasWork reports whether there is any works for give execution segment.
+func (carc ConstantArrivalRateConfig) HasWork(es *lib.ExecutionSegment) bool {
+	return carc.GetMaxVUs(es) > 0
+}
+
 // ConstantArrivalRate tries to execute a specific number of iterations for a
 // specific period.
 type ConstantArrivalRate struct {

lib/executor/constant_looping_vus.go

@@ -116,6 +116,11 @@ func (clvc ConstantLoopingVUsConfig) GetExecutionRequirements(es *lib.ExecutionS
 	}
 }
 
+// HasWork reports whether there is any works for give execution segment.
+func (clvc ConstantLoopingVUsConfig) HasWork(es *lib.ExecutionSegment) bool {
+	return clvc.GetVUs(es) > 0
+}
+
 // NewExecutor creates a new ConstantLoopingVUs executor
 func (clvc ConstantLoopingVUsConfig) NewExecutor(es *lib.ExecutionState, logger *logrus.Entry) (lib.Executor, error) {
 	return ConstantLoopingVUs{

lib/executor/externally_controlled.go

@@ -180,6 +180,11 @@ func (mec ExternallyControlledConfig) NewExecutor(es *lib.ExecutionState, logger
 	}, nil
 }
 
+// HasWork reports whether there is any works for give execution segment.
+func (mec ExternallyControlledConfig) HasWork(es *lib.ExecutionSegment) bool {
+	return es.Scale(mec.MaxVUs.Int64) > 0
+}
+
 type pauseEvent struct {
 	isPaused bool
 	err      chan error

lib/executor/per_vu_iterations.go

@@ -134,6 +134,11 @@ func (pvic PerVUIterationsConfig) NewExecutor(
 	}, nil
 }
 
+// HasWork reports whether there is any works for give execution segment.
+func (pvic PerVUIterationsConfig) HasWork(es *lib.ExecutionSegment) bool {
+	return pvic.GetVUs(es) > 0 && pvic.GetIterations() > 0
+}
+
 // PerVUIterations executes a specific number of iterations with each VU.
 type PerVUIterations struct {
 	*BaseExecutor

lib/executor/shared_iterations.go

@@ -148,6 +148,11 @@ type SharedIterations struct {
 // Make sure we implement the lib.Executor interface.
 var _ lib.Executor = &SharedIterations{}
 
+// HasWork reports whether there is any works for give execution segment.
+func (sic SharedIterationsConfig) HasWork(es *lib.ExecutionSegment) bool {
+	return sic.GetVUs(es) > 0 && sic.GetIterations(es) > 0
+}
+
 // Run executes a specific total number of iterations, which are all shared by
 // the configured VUs.
 func (si SharedIterations) Run(ctx context.Context, out chan<- stats.SampleContainer) (err error) {

lib/executor/variable_arrival_rate.go

@@ -260,6 +260,11 @@ func (varc VariableArrivalRateConfig) NewExecutor(
 	}, nil
 }
 
+// HasWork reports whether there is any works for give execution segment.
+func (varc VariableArrivalRateConfig) HasWork(es *lib.ExecutionSegment) bool {
+	return varc.GetMaxVUs(es) > 0
+}
+
 // VariableArrivalRate tries to execute a specific number of iterations for a
 // specific period.
 //TODO: combine with the ConstantArrivalRate?

lib/executor/variable_looping_vus.go

@@ -460,6 +460,11 @@ func (vlvc VariableLoopingVUsConfig) NewExecutor(es *lib.ExecutionState, logger
 	}, nil
 }
 
+// HasWork reports whether there is any works for give execution segment.
+func (vlvc VariableLoopingVUsConfig) HasWork(es *lib.ExecutionSegment) bool {
+	return lib.GetMaxPlannedVUs(vlvc.GetExecutionRequirements(es)) > 0
+}
+
 // VariableLoopingVUs handles the old "stages" execution configuration - it
 // loops iterations with a variable number of VUs for the sum of all of the
 // specified stages' duration.

lib/executors.go

@@ -103,6 +103,9 @@ type ExecutorConfig interface {
 	GetDescription(es *ExecutionSegment) string
 
 	NewExecutor(*ExecutionState, *logrus.Entry) (Executor, error)
+
+	// HasWork reports whether there is any work to do with given segment.
+	HasWork(*ExecutionSegment) bool
 }
 
 // InitVUFunc is just a shorthand so we don't have to type the function