lib/exec: Avoid to spawn a new goroutine for every iteration

Create a dedicated goroutine to process iterations
when a new ActiveVU is created for the Ramping arrival rate.

It changes the previous behaviour where instead
a new goroutine was created for every new iteration.

The previous solution was impacting the expected rate (iterations/s)
using a number of PreAllocatedVUs around thousand.
Added a Benchmark to check the rate with a set of PreAllocatedVUs cases.

Closes #1944

lib/executor/common_test.go

@@ -42,7 +42,7 @@ func simpleRunner(vuFn func(context.Context) error) lib.Runner {
 	}
 }
 
-func setupExecutor(t *testing.T, config lib.ExecutorConfig, es *lib.ExecutionState, runner lib.Runner) (
+func setupExecutor(t testing.TB, config lib.ExecutorConfig, es *lib.ExecutionState, runner lib.Runner) (
 	context.Context, context.CancelFunc, lib.Executor, *testutils.SimpleLogrusHook,
 ) {
 	ctx, cancel := context.WithCancel(context.Background())

lib/executor/ramping_arrival_rate.go

@@ -316,25 +316,32 @@ func (varr RampingArrivalRate) Run(parentCtx context.Context, out chan<- stats.S
 	returnedVUs := make(chan struct{})
 	startTime, maxDurationCtx, regDurationCtx, cancel := getDurationContexts(parentCtx, duration, gracefulStop)
 
+	vusPool := newActiveVUPool()
+
 	defer func() {
+		vusPool.Close()
 		// Make sure all VUs aren't executing iterations anymore, for the cancel()
 		// below to deactivate them.
 		<-returnedVUs
 		cancel()
 		activeVUsWg.Wait()
 	}()
-	activeVUs := make(chan lib.ActiveVU, maxVUs)
+
 	activeVUsCount := uint64(0)
 
 	returnVU := func(u lib.InitializedVU) {
 		varr.executionState.ReturnVU(u, true)
 		activeVUsWg.Done()
 	}
+
+	runIterationBasic := getIterationRunner(varr.executionState, varr.logger)
 	activateVU := func(initVU lib.InitializedVU) lib.ActiveVU {
 		activeVUsWg.Add(1)
 		activeVU := initVU.Activate(getVUActivationParams(maxDurationCtx, varr.config.BaseConfig, returnVU))
 		varr.executionState.ModCurrentlyActiveVUsCount(+1)
 		atomic.AddUint64(&activeVUsCount, 1)
+
+		vusPool.AddVU(maxDurationCtx, activeVU, runIterationBasic)
 		return activeVU
 	}
 
@@ -343,13 +350,7 @@ func (varr RampingArrivalRate) Run(parentCtx context.Context, out chan<- stats.S
 	defer close(makeUnplannedVUCh)
 	go func() {
 		defer close(returnedVUs)
-		defer func() {
-			// this is done here as to not have an unplannedVU in the middle of initialization when
-			// starting to return activeVUs
-			for i := uint64(0); i < atomic.LoadUint64(&activeVUsCount); i++ {
-				<-activeVUs
-			}
-		}()
+
 		for range makeUnplannedVUCh {
 			varr.logger.Debug("Starting initialization of an unplanned VU...")
 			initVU, err := varr.executionState.GetUnplannedVU(maxDurationCtx, varr.logger)
@@ -358,7 +359,7 @@ func (varr RampingArrivalRate) Run(parentCtx context.Context, out chan<- stats.S
 				varr.logger.WithError(err).Error("Error while allocating unplanned VU")
 			} else {
 				varr.logger.Debug("The unplanned VU finished initializing successfully!")
-				activeVUs <- activateVU(initVU)
+				activateVU(initVU)
 			}
 		}
 	}()
@@ -369,7 +370,7 @@ func (varr RampingArrivalRate) Run(parentCtx context.Context, out chan<- stats.S
 		if err != nil {
 			return err
 		}
-		activeVUs <- activateVU(initVU)
+		activateVU(initVU)
 	}
 
 	tickerPeriod := int64(startTickerPeriod.Duration)
@@ -380,9 +381,8 @@ func (varr RampingArrivalRate) Run(parentCtx context.Context, out chan<- stats.S
 	progressFn := func() (float64, []string) {
 		currActiveVUs := atomic.LoadUint64(&activeVUsCount)
 		currentTickerPeriod := atomic.LoadInt64(&tickerPeriod)
-		vusInBuffer := uint64(len(activeVUs))
 		progVUs := fmt.Sprintf(vusFmt+"/"+vusFmt+" VUs",
-			currActiveVUs-vusInBuffer, currActiveVUs)
+			vusPool.Running(), currActiveVUs)
 
 		itersPerSec := 0.0
 		if currentTickerPeriod > 0 {
@@ -408,12 +408,6 @@ func (varr RampingArrivalRate) Run(parentCtx context.Context, out chan<- stats.S
 	go trackProgress(parentCtx, maxDurationCtx, regDurationCtx, varr, progressFn)
 
 	regDurationDone := regDurationCtx.Done()
-	runIterationBasic := getIterationRunner(varr.executionState, varr.logger)
-	runIteration := func(vu lib.ActiveVU) {
-		runIterationBasic(maxDurationCtx, vu)
-		activeVUs <- vu
-	}
-
 	timer := time.NewTimer(time.Hour)
 	start := time.Now()
 	ch := make(chan time.Duration, 10) // buffer 10 iteration times ahead
@@ -439,12 +433,10 @@ func (varr RampingArrivalRate) Run(parentCtx context.Context, out chan<- stats.S
 			}
 		}
 
-		select {
-		case vu := <-activeVUs: // ideally, we get the VU from the buffer without any issues
-			go runIteration(vu) //TODO: refactor so we dont spin up a goroutine for each iteration
+		if vusPool.TryRunIteration() {
 			continue
-		default: // no free VUs currently available
 		}
+
 		// Since there aren't any free VUs available, consider this iteration
 		// dropped - we aren't going to try to recover it, but
 
@@ -471,3 +463,57 @@ func (varr RampingArrivalRate) Run(parentCtx context.Context, out chan<- stats.S
 	}
 	return nil
 }
+
+// activeVUPool controls the activeVUs
+// executing the received requests for iterations.
+type activeVUPool struct {
+	iterations chan struct{}
+	running    uint64
+	wg         sync.WaitGroup
+}
+
+// newActiveVUPool returns an activeVUPool.
+func newActiveVUPool() *activeVUPool {
+	return &activeVUPool{
+		iterations: make(chan struct{}),
+	}
+}
+
+// Run invokes a request to execute a new iteration.
+// When there aren'p *available VUs to process the request
+// then false is returned.
+func (p *activeVUPool) TryRunIteration() bool {
+	select {
+	case p.iterations <- struct{}{}:
+		return true
+	default:
+		return false
+	}
+}
+
+// Running returns the number of the currently running VUs.
+func (p *activeVUPool) Running() uint64 {
+	return atomic.LoadUint64(&p.running)
+}
+
+// AddVU adds the active VU to the pool of VUs for handling the incoming requests.
+// When a new request is accepted the runfn function is executed.
+func (p *activeVUPool) AddVU(ctx context.Context, avu lib.ActiveVU, runfn func(context.Context, lib.ActiveVU) bool) {
+	p.wg.Add(1)
+	go func() {
+		defer p.wg.Done()
+
+		for range p.iterations {
+			atomic.AddUint64(&p.running, uint64(1))
+			runfn(ctx, avu)
+			atomic.AddUint64(&p.running, ^uint64(0))
+		}
+	}()
+}
+
+// Close stops the pool from accepting requests
+// then it will wait thap *all the on-going iterations will be completed.
+func (p *activeVUPool) Close() {
+	close(p.iterations)
+	p.wg.Wait()
+}

lib/executor/ramping_arrival_rate_test.go

@@ -136,8 +136,8 @@ func TestRampingArrivalRateRunUnplannedVUs(t *testing.T) {
 	require.NoError(t, err)
 	es := lib.NewExecutionState(lib.Options{}, et, 1, 3)
 	var count int64
-	var ch = make(chan struct{})  // closed when new unplannedVU is started and signal to get to next iterations
-	var ch2 = make(chan struct{}) // closed when a second iteration was started on an old VU in order to test it won't start a second unplanned VU in parallel or at all
+	ch := make(chan struct{})  // closed when new unplannedVU is started and signal to get to next iterations
+	ch2 := make(chan struct{}) // closed when a second iteration was started on an old VU in order to test it won't start a second unplanned VU in parallel or at all
 	runner := simpleRunner(func(ctx context.Context) error {
 		cur := atomic.AddInt64(&count, 1)
 		if cur == 1 {
@@ -148,7 +148,7 @@ func TestRampingArrivalRateRunUnplannedVUs(t *testing.T) {
 
 		return nil
 	})
-	var ctx, cancel, executor, logHook = setupExecutor(
+	ctx, cancel, executor, logHook := setupExecutor(
 		t, &RampingArrivalRateConfig{
 			TimeUnit: types.NullDurationFrom(time.Second),
 			Stages: []Stage{
@@ -200,7 +200,7 @@ func TestRampingArrivalRateRunCorrectRateWithSlowRate(t *testing.T) {
 	require.NoError(t, err)
 	es := lib.NewExecutionState(lib.Options{}, et, 1, 3)
 	var count int64
-	var ch = make(chan struct{}) // closed when new unplannedVU is started and signal to get to next iterations
+	ch := make(chan struct{}) // closed when new unplannedVU is started and signal to get to next iterations
 	runner := simpleRunner(func(ctx context.Context) error {
 		cur := atomic.AddInt64(&count, 1)
 		if cur == 1 {
@@ -209,7 +209,7 @@ func TestRampingArrivalRateRunCorrectRateWithSlowRate(t *testing.T) {
 
 		return nil
 	})
-	var ctx, cancel, executor, logHook = setupExecutor(
+	ctx, cancel, executor, logHook := setupExecutor(
 		t, &RampingArrivalRateConfig{
 			TimeUnit: types.NullDurationFrom(time.Second),
 			Stages: []Stage{
@@ -223,7 +223,7 @@ func TestRampingArrivalRateRunCorrectRateWithSlowRate(t *testing.T) {
 		},
 		es, runner)
 	defer cancel()
-	var engineOut = make(chan stats.SampleContainer, 1000)
+	engineOut := make(chan stats.SampleContainer, 1000)
 	es.SetInitVUFunc(func(_ context.Context, logger *logrus.Entry) (lib.InitializedVU, error) {
 		t.Log("init")
 		cur := atomic.LoadInt64(&count)
@@ -243,6 +243,109 @@ func TestRampingArrivalRateRunCorrectRateWithSlowRate(t *testing.T) {
 	assert.Equal(t, int64(2), es.GetInitializedVUsCount())
 }
 
+func TestRampingArrivalRateRunGracefulStop(t *testing.T) {
+	t.Parallel()
+	et, err := lib.NewExecutionTuple(nil, nil)
+	require.NoError(t, err)
+	es := lib.NewExecutionState(lib.Options{}, et, 10, 10)
+
+	runner := simpleRunner(func(ctx context.Context) error {
+		time.Sleep(5 * time.Second)
+		return nil
+	})
+	ctx, cancel, executor, _ := setupExecutor(
+		t, &RampingArrivalRateConfig{
+			TimeUnit: types.NullDurationFrom(1 * time.Second),
+			Stages: []Stage{
+				{
+					Duration: types.NullDurationFrom(2 * time.Second),
+					Target:   null.IntFrom(10),
+				},
+			},
+			StartRate:       null.IntFrom(10),
+			PreAllocatedVUs: null.IntFrom(10),
+			MaxVUs:          null.IntFrom(10),
+			BaseConfig: BaseConfig{
+				GracefulStop: types.NullDurationFrom(5 * time.Second),
+			},
+		},
+		es, runner)
+	defer cancel()
+
+	engineOut := make(chan stats.SampleContainer, 1000)
+	defer close(engineOut)
+
+	err = executor.Run(ctx, engineOut)
+	assert.NoError(t, err)
+	assert.Equal(t, int64(0), es.GetCurrentlyActiveVUsCount())
+	assert.Equal(t, int64(10), es.GetInitializedVUsCount())
+	assert.Equal(t, uint64(10), es.GetFullIterationCount())
+}
+
+func BenchmarkRampingArrivalRateRun(b *testing.B) {
+	tests := []struct {
+		prealloc null.Int
+	}{
+		{prealloc: null.IntFrom(10)},
+		{prealloc: null.IntFrom(100)},
+		{prealloc: null.IntFrom(1e3)},
+		{prealloc: null.IntFrom(10e3)},
+	}
+
+	for _, tc := range tests {
+		b.Run(fmt.Sprintf("VUs%d", tc.prealloc.ValueOrZero()), func(b *testing.B) {
+			engineOut := make(chan stats.SampleContainer, 1000)
+			defer close(engineOut)
+			go func() {
+				for range engineOut {
+					// discard
+				}
+			}()
+
+			es := lib.NewExecutionState(lib.Options{}, mustNewExecutionTuple(nil, nil), uint64(tc.prealloc.Int64), uint64(tc.prealloc.Int64))
+
+			var count int64
+			runner := simpleRunner(func(ctx context.Context) error {
+				atomic.AddInt64(&count, 1)
+				return nil
+			})
+
+			// an high target to get the highest rate
+			target := int64(1e9)
+
+			ctx, cancel, executor, _ := setupExecutor(
+				b, &RampingArrivalRateConfig{
+					TimeUnit: types.NullDurationFrom(1 * time.Second),
+					Stages: []Stage{
+						{
+							Duration: types.NullDurationFrom(0),
+							Target:   null.IntFrom(target),
+						},
+						{
+							Duration: types.NullDurationFrom(5 * time.Second),
+							Target:   null.IntFrom(target),
+						},
+					},
+					PreAllocatedVUs: tc.prealloc,
+					MaxVUs:          tc.prealloc,
+				},
+				es, runner)
+			defer cancel()
+
+			b.ResetTimer()
+			start := time.Now()
+
+			err := executor.Run(ctx, engineOut)
+			took := time.Since(start)
+			assert.NoError(b, err)
+
+			iterations := float64(atomic.LoadInt64(&count))
+			b.ReportMetric(0, "ns/op")
+			b.ReportMetric(iterations/took.Seconds(), "iterations/s")
+		})
+	}
+}
+
 func mustNewExecutionTuple(seg *lib.ExecutionSegment, seq *lib.ExecutionSegmentSequence) *lib.ExecutionTuple {
 	et, err := lib.NewExecutionTuple(seg, seq)
 	if err != nil {