Fix constant-arrival-rate executor hanging

See #1368 (comment)

lib/executor/constant_arrival_rate.go

@@ -207,9 +207,6 @@ func (car ConstantArrivalRate) Run(ctx context.Context, out chan<- stats.SampleC
 	tickerPeriod := time.Duration(getTickerPeriod(arrivalRate).Duration)
 	arrivalRatePerSec, _ := getArrivalRatePerSec(arrivalRate).Float64()
 
-	startTime, maxDurationCtx, regDurationCtx, cancel := getDurationContexts(ctx, duration, gracefulStop)
-	defer cancel()
-
 	// Make sure the log and the progress bar have accurate information
 	car.logger.WithFields(logrus.Fields{
 		"maxVUs": maxVUs, "preAllocatedVUs": preAllocatedVUs, "duration": duration,
@@ -222,12 +219,16 @@ func (car ConstantArrivalRate) Run(ctx context.Context, out chan<- stats.SampleC
 
 	activeVUsWg := &sync.WaitGroup{}
 	defer activeVUsWg.Wait()
-	// Make sure we put planned and unplanned VUs back in the global buffer
+
+	startTime, maxDurationCtx, regDurationCtx, cancel := getDurationContexts(ctx, duration, gracefulStop)
+	defer cancel()
+
+	// Make sure all VUs aren't executing iterations anymore, for the cancel()
+	// above to deactivate them.
 	defer func() {
-		currActiveVUs := atomic.LoadUint64(&activeVUsCount)
-		for i := uint64(0); i < currActiveVUs; i++ {
-			vu := <-activeVUs
-			car.executionState.ReturnVU(vu.(lib.InitializedVU), false)
+		// activeVUsCount is modified only in the loop below, which is done here
+		for i := uint64(0); i < activeVUsCount; i++ {
+			<-activeVUs
 		}
 	}()
 
@@ -279,13 +280,9 @@ func (car ConstantArrivalRate) Run(ctx context.Context, out chan<- stats.SampleC
 	car.progress.Modify(pb.WithProgress(progresFn))
 	go trackProgress(ctx, maxDurationCtx, regDurationCtx, &car, progresFn)
 
-	regDurationDone := regDurationCtx.Done()
 	runIterationBasic := getIterationRunner(car.executionState, car.logger)
 	runIteration := func(vu lib.ActiveVU) {
-		ctx, cancel := context.WithCancel(maxDurationCtx)
-		defer cancel()
-
-		runIterationBasic(ctx, vu)
+		runIterationBasic(maxDurationCtx, vu)
 		activeVUs <- vu
 	}
 
@@ -302,7 +299,7 @@ func (car ConstantArrivalRate) Run(ctx context.Context, out chan<- stats.SampleC
 			)).Duration)
 
 	for li, gi := 0, start; ; li, gi = li+1, gi+offsets[li%len(offsets)] {
-		var t = notScaledTickerPeriod*time.Duration(gi) - time.Since(startTime)
+		t := notScaledTickerPeriod*time.Duration(gi) - time.Since(startTime)
 		timer.Reset(t)
 		select {
 		case <-timer.C:
@@ -312,7 +309,7 @@ func (car ConstantArrivalRate) Run(ctx context.Context, out chan<- stats.SampleC
 				go runIteration(vu)
 			default:
 				if remainingUnplannedVUs == 0 {
-					//TODO: emit an error metric?
+					// TODO: emit an error metric?
 					car.logger.Warningf("Insufficient VUs, reached %d active VUs and cannot allocate more", maxVUs)
 					break
 				}
@@ -323,7 +320,7 @@ func (car ConstantArrivalRate) Run(ctx context.Context, out chan<- stats.SampleC
 				remainingUnplannedVUs--
 				go runIteration(activateVU(initVU))
 			}
-		case <-regDurationDone:
+		case <-regDurationCtx.Done():
 			return nil
 		}
 	}