diff --git a/change.patch b/change.patch new file mode 100644 index 000000000000..c3a9f94bbed1 --- /dev/null +++ b/change.patch @@ -0,0 +1,60 @@ +diff --git a/lib/executor/ramping_vus.go b/lib/executor/ramping_vus.go +index 9faea830..75df7313 100644 +--- a/lib/executor/ramping_vus.go ++++ b/lib/executor/ramping_vus.go +@@ -543,8 +543,13 @@ func (vlv RampingVUs) Run(ctx context.Context, _ chan<- stats.SampleContainer, _ + for i := uint64(0); i < runState.maxVUs; i++ { + go runState.vuHandles[i].runLoopsIfPossible(runState.runIteration) + } +- runState.handleVUs(ctx) +- go runState.handleRemainingVUs(ctx) ++ ++ var ( ++ handleNewMaxAllowedVUs = runState.maxAllowedVUsHandlerStrategy() ++ handleNewScheduledVUs = runState.scheduledVUsHandlerStrategy() ++ ) ++ runState.handleVUs(ctx, handleNewMaxAllowedVUs, handleNewScheduledVUs) ++ go runState.handleRemainingVUs(ctx, handleNewMaxAllowedVUs) + + return nil + } +@@ -606,17 +611,16 @@ func (rs rampingVUsRunState) populateVUHandles(ctx context.Context, cancel func( + } + } + +-func (rs rampingVUsRunState) handleVUs(ctx context.Context) { ++func (rs rampingVUsRunState) handleVUs( ++ ctx context.Context, ++ handleNewMaxAllowedVUs, handleNewScheduledVUs func(lib.ExecutionStep), ++) { + // iterate over rawSteps and gracefulSteps in order by TimeOffset + // giving rawSteps precedence. + // we stop iterating once rawSteps are over as we need to run the remaining + // gracefulSteps concurrently while waiting for VUs to stop in order to not wait until + // the end of gracefulStop (= maxDuration-regularDuration) timeouts +- var ( +- handleNewMaxAllowedVUs = rs.maxAllowedVUsHandlerStrategy() +- handleNewScheduledVUs = rs.scheduledVUsHandlerStrategy() +- wait = waiter(ctx, rs.started) +- ) ++ wait := waiter(ctx, rs.started) + for i, j := 0, 0; i != len(rs.rawSteps); { + r, g := rs.rawSteps[i], rs.gracefulSteps[j] + if g.TimeOffset < r.TimeOffset { +@@ -635,11 +639,11 @@ func (rs rampingVUsRunState) handleVUs(ctx context.Context) { + } + } + +-func (rs rampingVUsRunState) handleRemainingVUs(ctx context.Context) { +- var ( +- handleNewMaxAllowedVUs = rs.maxAllowedVUsHandlerStrategy() +- wait = waiter(ctx, rs.started) +- ) ++func (rs rampingVUsRunState) handleRemainingVUs( ++ ctx context.Context, ++ handleNewMaxAllowedVUs func(lib.ExecutionStep), ++) { ++ wait := waiter(ctx, rs.started) + for _, s := range rs.gracefulSteps { + if wait(s.TimeOffset) { + return diff --git a/lib/executor/ramping_vus.go b/lib/executor/ramping_vus.go index 9faea830fe10..b0b5a6e3d6b4 100644 --- a/lib/executor/ramping_vus.go +++ b/lib/executor/ramping_vus.go @@ -543,9 +543,28 @@ func (vlv RampingVUs) Run(ctx context.Context, _ chan<- stats.SampleContainer, _ for i := uint64(0); i < runState.maxVUs; i++ { go runState.vuHandles[i].runLoopsIfPossible(runState.runIteration) } - runState.handleVUs(ctx) - go runState.handleRemainingVUs(ctx) + var ( + handleNewMaxAllowedVUs = runState.maxAllowedVUsHandlerStrategy() + handleNewScheduledVUs = runState.scheduledVUsHandlerStrategy() + ) + remainingGracefulSteps := runState.handleVUs( + ctx, + handleNewMaxAllowedVUs, + handleNewScheduledVUs, + ) + // Run the remaining gracefulSteps concurrently before the gracefulStop + // timeout period stops VUs. + // + // This way we will have run the gracefulSteps at the same time while + // waiting for the VUs to finish. + // + // (gracefulStop = maxDuration-regularDuration) + go runState.handleRemainingVUs( + ctx, + handleNewMaxAllowedVUs, + remainingGracefulSteps, + ) return nil } @@ -606,41 +625,43 @@ func (rs rampingVUsRunState) populateVUHandles(ctx context.Context, cancel func( } } -func (rs rampingVUsRunState) handleVUs(ctx context.Context) { - // iterate over rawSteps and gracefulSteps in order by TimeOffset - // giving rawSteps precedence. - // we stop iterating once rawSteps are over as we need to run the remaining - // gracefulSteps concurrently while waiting for VUs to stop in order to not wait until - // the end of gracefulStop (= maxDuration-regularDuration) timeouts - var ( - handleNewMaxAllowedVUs = rs.maxAllowedVUsHandlerStrategy() - handleNewScheduledVUs = rs.scheduledVUsHandlerStrategy() - wait = waiter(ctx, rs.started) - ) - for i, j := 0, 0; i != len(rs.rawSteps); { +// handleVUs iterates over rawSteps and gracefulSteps in order according to +// their TimeOffsets, prioritizing rawSteps. It stops iterating once rawSteps +// are over. And it returns the number of remaining gracefulSteps that it +// didn't run yet. +func (rs rampingVUsRunState) handleVUs( + ctx context.Context, + handleNewMaxAllowedVUs, handleNewScheduledVUs func(lib.ExecutionStep), +) (remaining int) { + wait := waiter(ctx, rs.started) + i, j := 0, 0 + for i != len(rs.rawSteps) { r, g := rs.rawSteps[i], rs.gracefulSteps[j] if g.TimeOffset < r.TimeOffset { j++ if wait(g.TimeOffset) { - return + break } handleNewMaxAllowedVUs(g) } else { i++ if wait(r.TimeOffset) { - return + break } handleNewScheduledVUs(r) } } + remaining = j + return remaining } -func (rs rampingVUsRunState) handleRemainingVUs(ctx context.Context) { - var ( - handleNewMaxAllowedVUs = rs.maxAllowedVUsHandlerStrategy() - wait = waiter(ctx, rs.started) - ) - for _, s := range rs.gracefulSteps { +func (rs rampingVUsRunState) handleRemainingVUs( + ctx context.Context, + handleNewMaxAllowedVUs func(lib.ExecutionStep), + remaining int, +) { + wait := waiter(ctx, rs.started) + for _, s := range rs.gracefulSteps[remaining:] { if wait(s.TimeOffset) { return }