diff --git a/bin/jmeter.properties b/bin/jmeter.properties index dd4dd7eaacd..bd0d1af9171 100644 --- a/bin/jmeter.properties +++ b/bin/jmeter.properties @@ -1327,6 +1327,16 @@ system.properties=system.properties # Disabled by default #testplan_validation.tpc_force_100_pct=false +#--------------------------------------------------------------------------- +# Open Model Thread Group configuration +#--------------------------------------------------------------------------- + +# Installation-wide cap on the number of threads running at the same time in every +# Open Model Thread Group that does not set its own concurrency(...) step. +# It is a safety net against running out of memory when the server responds slower +# than arrivals are scheduled. Values below one mean unlimited (the default). +#openmodel.max_concurrency=0 + #--------------------------------------------------------------------------- # Think Time configuration #--------------------------------------------------------------------------- diff --git a/src/core/src/main/kotlin/org/apache/jmeter/threads/openmodel/ConcurrencyGate.kt b/src/core/src/main/kotlin/org/apache/jmeter/threads/openmodel/ConcurrencyGate.kt new file mode 100644 index 00000000000..2f3ccdc120d --- /dev/null +++ b/src/core/src/main/kotlin/org/apache/jmeter/threads/openmodel/ConcurrencyGate.kt @@ -0,0 +1,177 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.jmeter.threads.openmodel + +import org.slf4j.LoggerFactory +import java.util.concurrent.TimeUnit +import java.util.concurrent.locks.ReentrantLock +import kotlin.concurrent.withLock + +/** + * Caps the number of threads that run at the same time in an [OpenModelThreadGroup]. + * + * The producer thread calls [acquire] before it clones the test plan, so a capped arrival waits for + * a free thread instead of allocating a fresh clone. This keeps the memory bounded by the limit and + * stops the group from piling threads onto a system that is already saturated. + * + * The limit is read from [concurrencyLimit] by the time elapsed since the schedule start, so it + * still increases while [acquire] is blocked. A blocked producer wakes up either when a thread + * finishes ([release]) or when the schedule raises the limit. + * + * @param concurrencyLimit per-schedule limit derived from `concurrency(...)` steps + * @param hardLimit installation-wide cap that always applies, or [Int.MAX_VALUE] for none + * @param testStartTime schedule start in [System.currentTimeMillis] units + * @param warnIntervalMillis minimum gap between saturation warnings + */ +internal class ConcurrencyGate( + private val concurrencyLimit: ConcurrencyLimit, + private val hardLimit: Int, + private val testStartTime: Long, + private val warnIntervalMillis: Long = TimeUnit.SECONDS.toMillis(10), +) { + private companion object { + private val log = LoggerFactory.getLogger(ConcurrencyGate::class.java) + } + + private val lock = ReentrantLock() + private val slotFreed = lock.newCondition() + private var running = 0 + + private var everSaturated = false + private var lastWarnMillis = Long.MIN_VALUE + + /** Number of arrivals that had to wait for a free thread. */ + @Volatile + var delayedArrivals: Long = 0 + private set + + /** Longest single arrival delay in milliseconds. */ + @Volatile + var maxDelayMillis: Long = 0 + private set + + /** Sum of all arrival delays in milliseconds. */ + @Volatile + var totalDelayMillis: Long = 0 + private set + + /** Number of arrivals dropped because the schedule ended before a thread became free. */ + @Volatile + var droppedArrivals: Long = 0 + private set + + /** True when no `concurrency(...)` step and no installation-wide cap apply. */ + val isUnlimited: Boolean get() = concurrencyLimit.isUnlimited && hardLimit == Int.MAX_VALUE + + private fun effectiveLimit(elapsedSeconds: Double): Int = + minOf(concurrencyLimit.limitAt(elapsedSeconds), hardLimit) + + /** + * Reserves a thread slot, blocking until one is free or [deadlineMillis] passes. The delay is + * recorded in the counters. + * @param deadlineMillis latest [System.currentTimeMillis] to keep waiting, usually the schedule + * end. An arrival that cannot start by then is dropped rather than stretching the test. + * @return true if a slot was reserved, false if the deadline passed first + * @throws InterruptedException if the producer is interrupted while waiting + */ + @Throws(InterruptedException::class) + fun acquire(deadlineMillis: Long): Boolean { + if (isUnlimited) { + return true + } + lock.lockInterruptibly() + try { + val entryMillis = System.currentTimeMillis() + var blocked = false + while (true) { + val now = System.currentTimeMillis() + val elapsedMillis = now - testStartTime + val limit = effectiveLimit(elapsedMillis / 1000.0) + if (running < limit) { + running++ + if (blocked) { + recordDelay(System.currentTimeMillis() - entryMillis) + } + return true + } + if (now >= deadlineMillis) { + droppedArrivals++ + return false + } + if (!blocked) { + blocked = true + maybeWarn(limit) + } + awaitSlotOrLimitChange(elapsedMillis, deadlineMillis) + } + } finally { + lock.unlock() + } + } + + /** Releases a slot reserved by [acquire] and wakes one blocked producer. */ + fun release() { + if (isUnlimited) { + return + } + lock.withLock { + running-- + slotFreed.signal() + } + } + + private fun awaitSlotOrLimitChange(elapsedMillis: Long, deadlineMillis: Long) { + // Wake up no later than the schedule deadline so a dropped arrival is not stuck forever + var waitMillis = deadlineMillis - (testStartTime + elapsedMillis) + val nextChangeSeconds = concurrencyLimit.nextChangeAfter(elapsedMillis / 1000.0) + if (nextChangeSeconds.isFinite()) { + waitMillis = minOf(waitMillis, (nextChangeSeconds * 1000).toLong() - elapsedMillis) + } + if (waitMillis > 0) { + slotFreed.await(waitMillis, TimeUnit.MILLISECONDS) + } + } + + private fun recordDelay(delayMillis: Long) { + delayedArrivals++ + totalDelayMillis += delayMillis + if (delayMillis > maxDelayMillis) { + maxDelayMillis = delayMillis + } + } + + private fun maybeWarn(limit: Int) { + val now = System.currentTimeMillis() + if (!everSaturated) { + everSaturated = true + lastWarnMillis = now + log.warn( + "Concurrency limit of {} thread(s) reached, further arrivals are delayed until a thread" + + " becomes free. Increase concurrency(...) or reduce the rate to keep an open model.", + limit + ) + } else if (now - lastWarnMillis >= warnIntervalMillis) { + lastWarnMillis = now + log.warn( + "Concurrency limit of {} thread(s) is still saturated, {} arrival(s) delayed so far" + + " (max delay {} ms).", + limit, delayedArrivals, maxDelayMillis + ) + } + } +} diff --git a/src/core/src/main/kotlin/org/apache/jmeter/threads/openmodel/ConcurrencyLimit.kt b/src/core/src/main/kotlin/org/apache/jmeter/threads/openmodel/ConcurrencyLimit.kt new file mode 100644 index 00000000000..a6ac61745d4 --- /dev/null +++ b/src/core/src/main/kotlin/org/apache/jmeter/threads/openmodel/ConcurrencyLimit.kt @@ -0,0 +1,94 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.jmeter.threads.openmodel + +/** + * The maximum number of concurrent threads over time, derived from the [ConcurrencyStep] entries + * of a [ThreadSchedule]. + * + * The limit is piecewise-constant: a [ConcurrencyStep] applies to the arrivals that follow it, + * until the next [ConcurrencyStep]. The value is looked up by the time elapsed since the schedule + * start rather than by the position of the arrivals generator, so the limit still increases even + * while the generator is blocked waiting for a free thread. + */ +internal class ConcurrencyLimit private constructor( + // Sorted by time. breakpointTimes[i] is the elapsed time in seconds when limits[i] takes effect. + private val breakpointTimes: DoubleArray, + private val limits: IntArray, +) { + /** True when the schedule has no [ConcurrencyStep], so the number of threads is unlimited. */ + val isUnlimited: Boolean get() = breakpointTimes.isEmpty() + + /** A point in time (seconds since the schedule start) where the limit changes to [limit]. */ + data class Change(val timeSeconds: Double, val limit: Int) + + /** The points where the limit changes, in schedule order. Empty when [isUnlimited]. */ + val changes: List get() = breakpointTimes.indices.map { Change(breakpointTimes[it], limits[it]) } + + /** + * The concurrency limit that applies at [elapsedSeconds] since the schedule start. + * Returns [Int.MAX_VALUE] before the first [ConcurrencyStep] or when the schedule has none. + */ + fun limitAt(elapsedSeconds: Double): Int { + var res = Int.MAX_VALUE + for (i in breakpointTimes.indices) { + if (breakpointTimes[i] > elapsedSeconds) { + break + } + res = limits[i] + } + return res + } + + /** + * The elapsed time in seconds of the earliest limit change strictly after [elapsedSeconds], + * or [Double.POSITIVE_INFINITY] if the limit never changes again. Lets a blocked producer wake + * up when the limit might increase, even if no thread has finished. + */ + fun nextChangeAfter(elapsedSeconds: Double): Double { + for (time in breakpointTimes) { + if (time > elapsedSeconds) { + return time + } + } + return Double.POSITIVE_INFINITY + } + + companion object { + fun of(schedule: ThreadSchedule): ConcurrencyLimit { + val times = mutableListOf() + val values = mutableListOf() + var time = 0.0 + for (step in schedule.steps) { + when (step) { + is ThreadScheduleStep.ConcurrencyStep -> + // Several concurrency steps at the same time offset: the last one wins + if (times.isNotEmpty() && times.last() == time) { + values[values.size - 1] = step.limit + } else { + times += time + values += step.limit + } + is ThreadScheduleStep.ArrivalsStep -> time += step.duration + is ThreadScheduleStep.RateStep -> Unit + } + } + return ConcurrencyLimit(times.toDoubleArray(), values.toIntArray()) + } + } +} diff --git a/src/core/src/main/kotlin/org/apache/jmeter/threads/openmodel/OpenModelThreadGroup.kt b/src/core/src/main/kotlin/org/apache/jmeter/threads/openmodel/OpenModelThreadGroup.kt index 45c0f6cef6b..681ee2fb1a9 100644 --- a/src/core/src/main/kotlin/org/apache/jmeter/threads/openmodel/OpenModelThreadGroup.kt +++ b/src/core/src/main/kotlin/org/apache/jmeter/threads/openmodel/OpenModelThreadGroup.kt @@ -27,6 +27,7 @@ import org.apache.jmeter.threads.JMeterThread import org.apache.jmeter.threads.JMeterThreadMonitor import org.apache.jmeter.threads.ListenerNotifier import org.apache.jmeter.threads.TestCompilerHelper +import org.apache.jmeter.util.JMeterUtils import org.apache.jorphan.collections.ListedHashTree import org.apiguardian.api.API import org.slf4j.LoggerFactory @@ -43,8 +44,9 @@ import kotlin.math.roundToLong /** * The thread group that emulates open model. - * Currently, threads are created on demand, every thread exists after completion, - * and the maximum number of threads is not limited. + * Threads are created on demand, and every thread exits after completion. + * The number of concurrent threads is unlimited by default, however, it can be capped with a + * `concurrency(...)` step in the schedule or with the `openmodel.max_concurrency` property. */ @GUIMenuSortOrder(1) @API(status = API.Status.EXPERIMENTAL, since = "5.5") @@ -85,6 +87,12 @@ public class OpenModelThreadGroup : */ private val houseKeepingThreadPool = Executors.newCachedThreadPool() + /** + * Installation-wide hard cap on the number of concurrent threads, applied even when the + * schedule has no `concurrency(...)` step. Values below one mean unlimited (the default). + */ + public const val MAX_CONCURRENCY_PROPERTY: String = "openmodel.max_concurrency" + private const val serialVersionUID: Long = 1L } @@ -96,6 +104,9 @@ public class OpenModelThreadGroup : private val threadStarterFuture = AtomicReference?>() private val activeThreads = ConcurrentHashMap>() + // Caps the number of concurrent threads. Created on test start. + private val concurrencyGate = AtomicReference() + override val schema: OpenModelThreadGroupSchema get() = OpenModelThreadGroupSchema @@ -123,6 +134,7 @@ public class OpenModelThreadGroup : private val executorService: ExecutorService, private val activeThreads: MutableMap>, private val gen: ThreadScheduleProcessGenerator, + private val concurrencyGate: ConcurrencyGate, private val jmeterThreadFactory: (threadNumber: Int) -> JMeterThread, ) : Runnable { override fun run() { @@ -140,6 +152,12 @@ public class OpenModelThreadGroup : sleep(nextDelay) } } + // Reserve a thread slot before cloning the test plan, so a capped arrival waits + // for a free thread instead of allocating a clone that would run out of memory. + if (!concurrencyGate.acquire(endTime)) { + // The schedule window ended while waiting for a free thread, so stop launching. + break + } val jmeterThread = jmeterThreadFactory(threadNumber++) jmeterThread.endTime = endTime activeThreads[jmeterThread] = executorService.submit { @@ -147,6 +165,7 @@ public class OpenModelThreadGroup : jmeterThread.run() } } + logConcurrencySummary() // If test schedule ends with a pause, then we need to wait for it val timeLeft = endTime - System.currentTimeMillis() if (timeLeft > 0) { @@ -183,6 +202,20 @@ public class OpenModelThreadGroup : executorService.shutdownNow() log.info("Thread starting done") } + + private fun logConcurrencySummary() { + val delayed = concurrencyGate.delayedArrivals + val dropped = concurrencyGate.droppedArrivals + if (delayed == 0L && dropped == 0L) { + return + } + log.warn( + "Concurrency limit affected the load: {} arrival(s) delayed (total {} ms, max {} ms)," + + " {} arrival(s) dropped after the schedule ended." + + " The achieved rate was below the target rate; raise concurrency(...) to reach it.", + delayed, concurrencyGate.totalDelayMillis, concurrencyGate.maxDelayMillis, dropped + ) + } } override fun recoverRunningVersion() { @@ -207,7 +240,11 @@ public class OpenModelThreadGroup : val testStartTime = this.startTime val executorService = Executors.newCachedThreadPool() this.executorService = executorService - val starter = ThreadsStarter(testStartTime, executorService, activeThreads, gen) { threadNumber -> + val hardLimit = JMeterUtils.getPropDefault(MAX_CONCURRENCY_PROPERTY, 0) + .let { if (it <= 0) Int.MAX_VALUE else it } + val gate = ConcurrencyGate(ConcurrencyLimit.of(parsedSchedule), hardLimit, testStartTime) + concurrencyGate.set(gate) + val starter = ThreadsStarter(testStartTime, executorService, activeThreads, gen, gate) { threadNumber -> val clonedTree = cloneTree(threadGroupTree) makeThread(engine, this, notifier, threadGroupIndex, threadNumber, clonedTree, variables) } @@ -226,6 +263,10 @@ public class OpenModelThreadGroup : override fun threadFinished(thread: JMeterThread?) { activeThreads.remove(thread) + // Every launched thread reserved a slot in the gate before it started and calls + // threadFinished exactly once, so release here regardless of the activeThreads state: + // a fast thread may finish before the starter records it in activeThreads. + concurrencyGate.get()?.release() } override fun addNewThread(delay: Int, engine: StandardJMeterEngine?): JMeterThread { @@ -238,6 +279,14 @@ public class OpenModelThreadGroup : override fun numberOfActiveThreads(): Int = activeThreads.size + /** Number of arrivals that had to wait for a free thread because of the concurrency limit. */ + @API(status = API.Status.EXPERIMENTAL, since = "6.0") + public fun getDelayedArrivals(): Long = concurrencyGate.get()?.delayedArrivals ?: 0 + + /** Number of arrivals dropped because the schedule ended before a thread became free. */ + @API(status = API.Status.EXPERIMENTAL, since = "6.0") + public fun getDroppedArrivals(): Long = concurrencyGate.get()?.droppedArrivals ?: 0 + override fun verifyThreadsStopped(): Boolean = executorService?.awaitTermination(0, TimeUnit.SECONDS) != false diff --git a/src/core/src/main/kotlin/org/apache/jmeter/threads/openmodel/ThreadScheduleProcessGenerator.kt b/src/core/src/main/kotlin/org/apache/jmeter/threads/openmodel/ThreadScheduleProcessGenerator.kt index a56295d0c2d..627e9f5ca1e 100644 --- a/src/core/src/main/kotlin/org/apache/jmeter/threads/openmodel/ThreadScheduleProcessGenerator.kt +++ b/src/core/src/main/kotlin/org/apache/jmeter/threads/openmodel/ThreadScheduleProcessGenerator.kt @@ -98,6 +98,8 @@ internal class ThreadScheduleProcessGenerator( arrivalStep = step break } + // Concurrency is enforced by ConcurrencyGate, so the arrivals generator ignores it + is ThreadScheduleStep.ConcurrencyStep -> Unit } } diff --git a/src/core/src/main/kotlin/org/apache/jmeter/threads/openmodel/gui/OpenModelThreadGroupGui.kt b/src/core/src/main/kotlin/org/apache/jmeter/threads/openmodel/gui/OpenModelThreadGroupGui.kt index 5188b1da8f3..dd1fb9d4a2a 100644 --- a/src/core/src/main/kotlin/org/apache/jmeter/threads/openmodel/gui/OpenModelThreadGroupGui.kt +++ b/src/core/src/main/kotlin/org/apache/jmeter/threads/openmodel/gui/OpenModelThreadGroupGui.kt @@ -81,8 +81,9 @@ public class OpenModelThreadGroupGui : AbstractThreadGroupGui() { private fun createPanel() = JPanel(MigLayout("fillx, wrap1", "[fill, grow]")).apply { - add(labelFor(scheduleStringEditor, "openmodelthreadgroup_schedule_string"), "grow 0, split 6") + add(labelFor(scheduleStringEditor, "openmodelthreadgroup_schedule_string"), "grow 0, split 7") add(templateButton("rate(1/min)"), "grow 0") + add(templateButton("concurrency(100)"), "grow 0") add(templateButton("random_arrivals(10 min)"), "grow 0") add(templateButton("pause(1 min)"), "grow 0") add(templateButton("/* comment */"), "grow 0") diff --git a/src/core/src/main/kotlin/org/apache/jmeter/threads/openmodel/gui/TargetRateChart.kt b/src/core/src/main/kotlin/org/apache/jmeter/threads/openmodel/gui/TargetRateChart.kt index f24150e1c6e..4cc865e55b8 100644 --- a/src/core/src/main/kotlin/org/apache/jmeter/threads/openmodel/gui/TargetRateChart.kt +++ b/src/core/src/main/kotlin/org/apache/jmeter/threads/openmodel/gui/TargetRateChart.kt @@ -17,6 +17,7 @@ package org.apache.jmeter.threads.openmodel.gui +import org.apache.jmeter.threads.openmodel.ConcurrencyLimit import org.apache.jmeter.threads.openmodel.ThreadSchedule import org.apache.jmeter.threads.openmodel.ThreadScheduleStep import org.apache.jmeter.threads.openmodel.asSeconds @@ -26,6 +27,9 @@ import org.jetbrains.letsPlot.batik.plot.component.DefaultPlotPanelBatik import org.jetbrains.letsPlot.commons.registration.Disposable import org.jetbrains.letsPlot.core.util.MonolithicCommon import org.jetbrains.letsPlot.geom.geomLine +import org.jetbrains.letsPlot.geom.geomStep +import org.jetbrains.letsPlot.gggrid +import org.jetbrains.letsPlot.intern.Plot import org.jetbrains.letsPlot.intern.toSpec import org.jetbrains.letsPlot.label.ggtitle import org.jetbrains.letsPlot.letsPlot @@ -52,14 +56,27 @@ public class TargetRateChart : JPanel() { } private var prevSteps: List? = null - private var prevTimes: DoubleArray? = null - private var prevRate: DoubleArray? = null public fun updateSchedule(threadSchedule: ThreadSchedule) { if (threadSchedule.steps == prevSteps) { return } prevSteps = threadSchedule.steps + val spec = buildSpec(threadSchedule) + components.forEach { + if (it is Disposable) { + it.dispose() + } + } + removeAll() + add(render(spec), BorderLayout.CENTER) + } + + /** + * Builds the lets-plot specification for the schedule. Kept separate from [updateSchedule] so it + * can be tested without instantiating the Swing/Batik rendering. + */ + internal fun buildSpec(threadSchedule: ThreadSchedule): MutableMap { val timeValues = mutableListOf() val rateValues = mutableListOf() var time = 0.0 @@ -82,56 +99,83 @@ public class TargetRateChart : JPanel() { addPoint = true time += step.duration } + // Concurrency is drawn as a separate panel below + is ThreadScheduleStep.ConcurrencyStep -> Unit } } if (addPoint) { timeValues += time rateValues += rate } - setData(timeValues.toDoubleArray(), rateValues.toDoubleArray()) - } - private fun setData(time: DoubleArray, rate: DoubleArray) { - if (time.contentEquals(prevTimes) && rate.contentEquals(prevRate)) { - return - } - prevTimes = time.copyOf() - prevRate = rate.copyOf() + val timeArray = timeValues.toDoubleArray() + val rateArray = rateValues.toDoubleArray() val timeScale = TimeUnit.SECONDS.toMillis(1).toDouble() - for (i in time.indices) { - time[i] *= timeScale + for (i in timeArray.indices) { + timeArray[i] *= timeScale } - val maxRate = rate.maxOrNull() ?: 0.0 + val maxRate = rateArray.maxOrNull() ?: 0.0 val rateUnit = rateUnitFor(maxRate) if (rateUnit != TimeUnit.SECONDS) { val scale = rateUnit.asSeconds - for (i in rate.indices) { - rate[i] *= scale + for (i in rateArray.indices) { + rateArray[i] *= scale } } - components.forEach { - if (it is Disposable) { - it.dispose() - } + val changes = ConcurrencyLimit.of(threadSchedule).changes + // A single limit that applies from the start is shown as a subtitle rather than a panel. + val constantLimit = changes.singleOrNull()?.takeIf { it.timeSeconds == 0.0 }?.limit + val drawPanel = changes.isNotEmpty() && constantLimit == null + + val ratePlot = createRateChart( + time = timeArray, + rate = rateArray, + rateUnit = rateUnit, + subtitle = constantLimit?.let { "Max threads: $it" }, + showTimeAxis = !drawPanel, + ) + return if (drawPanel) { + val concurrencyPlot = createConcurrencyChart(changes, time * timeScale, timeScale) + // Stack the two panels; sharex aligns the time axes so they line up vertically + LinkedHashMap(gggrid(listOf(ratePlot, concurrencyPlot), ncol = 1, sharex = "all", align = true).toSpec()) + } else { + ratePlot.toSpec() } - removeAll() - add(createChart(time = time, rate = rate, rateUnit = rateUnit), BorderLayout.CENTER) } - private fun createChart(time: DoubleArray, rate: DoubleArray, rateUnit: TimeUnit): JComponent { - val data = mapOf( - "time" to time, - "rate" to rate - ) - val plot = letsPlot(data) + geomLine { x = "time"; y = "rate" } + + private fun createRateChart( + time: DoubleArray, + rate: DoubleArray, + rateUnit: TimeUnit, + subtitle: String?, + showTimeAxis: Boolean, + ): Plot { + val data = mapOf("time" to time, "rate" to rate) + var plot = letsPlot(data) + geomLine { x = "time"; y = "rate" } + scaleXTime("Time since test start", expand = listOf(0, 0)) + - ggtitle("Target load rate per " + rateUnit.name.lowercase().removeSuffix("s")) + + ggtitle("Target load rate per " + rateUnit.name.lowercase().removeSuffix("s"), subtitle) + theme(axisTitleY = "blank") + if (!showTimeAxis) { + // The concurrency panel below carries the shared time axis + plot += theme(axisTitleX = "blank", axisTextX = "blank") + } + return plot + } - val rawSpec = plot.toSpec() - val processedSpec = MonolithicCommon.processRawSpecs(rawSpec, frontendOnly = false) + private fun createConcurrencyChart(changes: List, endTimeMillis: Double, timeScale: Double): Plot { + val time = DoubleArray(changes.size + 1) { if (it < changes.size) changes[it].timeSeconds * timeScale else endTimeMillis } + val limit = DoubleArray(changes.size + 1) { changes[if (it < changes.size) it else changes.size - 1].limit.toDouble() } + val data = mapOf("time" to time, "threads" to limit) + // geomStep, not geomLine: the limit is piecewise-constant and jumps between windows + return letsPlot(data) + geomStep { x = "time"; y = "threads" } + + scaleXTime("Time since test start", expand = listOf(0, 0)) + + ggtitle("Max concurrent threads") + + theme(axisTitleY = "blank") + } + private fun render(spec: MutableMap): JComponent { + val processedSpec = MonolithicCommon.processRawSpecs(spec, frontendOnly = false) return DefaultPlotPanelBatik( processedSpec = processedSpec, preserveAspectRatio = false, diff --git a/src/core/src/main/kotlin/org/apache/jmeter/threads/openmodel/scheduleParser.kt b/src/core/src/main/kotlin/org/apache/jmeter/threads/openmodel/scheduleParser.kt index 965cc4d47fc..913ae932a34 100644 --- a/src/core/src/main/kotlin/org/apache/jmeter/threads/openmodel/scheduleParser.kt +++ b/src/core/src/main/kotlin/org/apache/jmeter/threads/openmodel/scheduleParser.kt @@ -36,6 +36,9 @@ import kotlin.jvm.Throws * * rate(1/sec) rate(2/sec) arrivals(1 min) rate(3/sec) rate(4/sec) arrivals(2 min) rate(5/sec)... * means 2/sec..3/sec during 1min, then 4/sec..5/sec during 2min + * + * concurrency(10) rate(2/sec) arrivals(1 min) concurrency(20) arrivals(2 min) + * means at most 10 concurrent threads during the first minute, and at most 20 during the next two */ @API(status = API.Status.EXPERIMENTAL, since = "5.5") public interface ThreadSchedule { @@ -84,6 +87,16 @@ public sealed interface ThreadScheduleStep { override fun toString(): String = "Rate(${FORMAT.format(rate)})" } + /** + * Limits the number of threads that execute the test plan at the same time. + * The limit applies to the arrivals that follow the step, so the arrivals that would exceed + * the limit wait for a free thread rather than start immediately. + */ + @API(status = API.Status.EXPERIMENTAL, since = "6.0") + public data class ConcurrencyStep(val limit: Int) : ThreadScheduleStep { + override fun toString(): String = "Concurrency($limit)" + } + @API(status = API.Status.EXPERIMENTAL, since = "5.5") public enum class ArrivalType { EVEN, RANDOM @@ -131,10 +144,11 @@ internal class ScheduleParser(private val schedule: String) { } // Variable is needed for formatting: https://youtrack.jetbrains.com/issue/KTIJ-19984 val step = parseRate() + ?: parseConcurrency() ?: parseArrivals("random_arrivals", ArrivalType.RANDOM) ?: parseArrivals("even_arrivals", ArrivalType.EVEN) ?: throwParseException( - "Unexpected input (expecting rate, random_arrivals, even_arrivals, or pause)" + "Unexpected input (expecting rate, concurrency, random_arrivals, even_arrivals, or pause)" ) steps += step if (step is ThreadScheduleStep.RateStep) { @@ -174,6 +188,21 @@ internal class ScheduleParser(private val schedule: String) { return ThreadScheduleStep.RateStep(rateValue / timeUnit.asSeconds) } + private fun parseConcurrency(): ThreadScheduleStep? { + if (!token?.image.equals("concurrency", ignoreCase = true)) { + return null + } + pos += 1 + consume(Tokenizer.OpenParenthesisToken) + val limit = consume("integer number for concurrency") + val limitValue = limit.image.toDouble() + if (limitValue < 1 || limitValue != Math.floor(limitValue)) { + throwParseException("concurrency must be a positive integer, got ${limit.image}") + } + consume(Tokenizer.CloseParenthesisToken) + return ThreadScheduleStep.ConcurrencyStep(limitValue.toInt()) + } + private fun parseArrivals(functionName: String, type: ArrivalType): ThreadScheduleStep? { if (!token?.image.equals(functionName, ignoreCase = true)) { return null diff --git a/src/core/src/test/kotlin/org/apache/jmeter/threads/openmodel/ConcurrencyGateTest.kt b/src/core/src/test/kotlin/org/apache/jmeter/threads/openmodel/ConcurrencyGateTest.kt new file mode 100644 index 00000000000..82f16448b11 --- /dev/null +++ b/src/core/src/test/kotlin/org/apache/jmeter/threads/openmodel/ConcurrencyGateTest.kt @@ -0,0 +1,93 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.jmeter.threads.openmodel + +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.Timeout +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicReference +import kotlin.concurrent.thread + +class ConcurrencyGateTest { + private val farFuture get() = System.currentTimeMillis() + TimeUnit.HOURS.toMillis(1) + + private fun gateFor(schedule: String, hardLimit: Int = Int.MAX_VALUE) = + ConcurrencyGate(ConcurrencyLimit.of(ThreadSchedule(schedule)), hardLimit, System.currentTimeMillis()) + + @Test + fun `unlimited gate never blocks`() { + val gate = gateFor("rate(1/sec) random_arrivals(1 hour)") + assertTrue(gate.isUnlimited, "isUnlimited") + repeat(1000) { + assertTrue(gate.acquire(farFuture), "acquire should not block when unlimited") + } + } + + @Test + fun `hard limit applies without a concurrency step`() { + val gate = gateFor("rate(1/sec) random_arrivals(1 hour)", hardLimit = 1) + assertFalse(gate.isUnlimited, "isUnlimited") + assertTrue(gate.acquire(farFuture), "first slot") + assertFalse(gate.acquire(System.currentTimeMillis() - 1), "second slot is dropped past the deadline") + assertEquals(1, gate.droppedArrivals) + } + + @Test + fun `arrival is dropped when the deadline passes`() { + val gate = gateFor("concurrency(1) rate(1/sec) random_arrivals(1 hour)") + assertTrue(gate.acquire(farFuture), "first slot is free") + assertFalse(gate.acquire(System.currentTimeMillis() - 1), "second slot cannot be reserved") + assertEquals(1, gate.droppedArrivals) + } + + @Test + fun `released slot unblocks a waiting producer`() { + val gate = gateFor("concurrency(1) rate(1/sec) random_arrivals(1 hour)") + assertTrue(gate.acquire(farFuture), "first slot") + val acquired = AtomicReference() + val waiter = thread { acquired.set(gate.acquire(farFuture)) } + Thread.sleep(200) + assertEquals(null, acquired.get(), "the waiter should still be blocked") + gate.release() + waiter.join(2000) + assertEquals(true, acquired.get(), "the waiter should acquire the freed slot") + assertTrue(gate.delayedArrivals >= 1, "the delayed arrival should be counted") + } + + @Test + @Timeout(5, unit = TimeUnit.SECONDS) + fun `blocked producer is interruptible`() { + val gate = gateFor("concurrency(1) rate(1/sec) random_arrivals(1 hour)") + assertTrue(gate.acquire(farFuture), "first slot") + val error = AtomicReference() + val waiter = thread { + try { + gate.acquire(farFuture) + } catch (t: InterruptedException) { + error.set(t) + } + } + Thread.sleep(200) + waiter.interrupt() + waiter.join(2000) + assertTrue(error.get() is InterruptedException, "acquire should throw InterruptedException when interrupted") + } +} diff --git a/src/core/src/test/kotlin/org/apache/jmeter/threads/openmodel/ConcurrencyLimitBehaviorTest.kt b/src/core/src/test/kotlin/org/apache/jmeter/threads/openmodel/ConcurrencyLimitBehaviorTest.kt new file mode 100644 index 00000000000..394686301f3 --- /dev/null +++ b/src/core/src/test/kotlin/org/apache/jmeter/threads/openmodel/ConcurrencyLimitBehaviorTest.kt @@ -0,0 +1,113 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.jmeter.threads.openmodel + +import org.apache.jmeter.engine.StandardJMeterEngine +import org.apache.jmeter.junit.JMeterTestCase +import org.apache.jmeter.samplers.AbstractSampler +import org.apache.jmeter.samplers.Entry +import org.apache.jmeter.samplers.SampleResult +import org.apache.jmeter.testelement.TestElementSchema +import org.apache.jmeter.testelement.TestPlan +import org.apache.jmeter.treebuilder.dsl.testTree +import org.apache.jorphan.test.JMeterSerialTest +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.Timeout +import java.time.Duration +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger + +/** + * Sampler that records how many threads run [sample] at the same time, so a test can assert the + * concurrency limit is respected. + */ +class ConcurrencyProbe : AbstractSampler() { + abstract class Schema : TestElementSchema() { + companion object INSTANCE : Schema() + + val sleep by long("ConcurrencyProbe.sleep") + } + + var sleepMillis by Schema.sleep + + override fun sample(e: Entry?): SampleResult { + val res = SampleResult() + res.sampleStart() + val concurrent = current.incrementAndGet() + maxConcurrency.accumulateAndGet(concurrent) { a, b -> maxOf(a, b) } + try { + Thread.sleep(sleepMillis) + res.responseCode = "OK" + } catch (t: InterruptedException) { + res.responseCode = "InterruptedException" + } finally { + current.decrementAndGet() + res.sampleEnd() + } + return res + } + + companion object { + val current = AtomicInteger() + val maxConcurrency = AtomicInteger() + + fun reset() { + current.set(0) + maxConcurrency.set(0) + } + } +} + +class ConcurrencyLimitBehaviorTest : JMeterTestCase(), JMeterSerialTest { + @Test + @Timeout(30, unit = TimeUnit.SECONDS) + fun `concurrency caps the number of simultaneous threads`() { + ConcurrencyProbe.reset() + var group: OpenModelThreadGroup? = null + + val tree = testTree { + TestPlan::class { + OpenModelThreadGroup::class { + group = this + name = "omtg" + // ~50 arrivals in one second, but only 2 threads may run at a time. + // The pause lets the delayed arrivals drain before the schedule ends. + scheduleString = "concurrency(2) rate(50/sec) random_arrivals(1 sec) pause(10 sec)" + ConcurrencyProbe::class { + sleepMillis = 200 + } + } + } + } + StandardJMeterEngine().apply { + configure(tree) + runTest() + awaitTermination(Duration.ofSeconds(30)) + } + + assertTrue( + ConcurrencyProbe.maxConcurrency.get() <= 2, + "At most 2 threads should run at a time, but observed ${ConcurrencyProbe.maxConcurrency.get()}" + ) + assertTrue( + group!!.getDelayedArrivals() > 0, + "Some arrivals should be delayed because the rate exceeds the concurrency limit" + ) + } +} diff --git a/src/core/src/test/kotlin/org/apache/jmeter/threads/openmodel/ConcurrencyLimitTest.kt b/src/core/src/test/kotlin/org/apache/jmeter/threads/openmodel/ConcurrencyLimitTest.kt new file mode 100644 index 00000000000..0174aa5001c --- /dev/null +++ b/src/core/src/test/kotlin/org/apache/jmeter/threads/openmodel/ConcurrencyLimitTest.kt @@ -0,0 +1,64 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.jmeter.threads.openmodel + +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test + +class ConcurrencyLimitTest { + @Test + fun `no concurrency step is unlimited`() { + val limit = ConcurrencyLimit.of(ThreadSchedule("rate(5/sec) random_arrivals(1 min)")) + assertTrue(limit.isUnlimited, "isUnlimited") + assertEquals(Int.MAX_VALUE, limit.limitAt(0.0)) + assertEquals(Double.POSITIVE_INFINITY, limit.nextChangeAfter(0.0)) + } + + @Test + fun `constant concurrency`() { + val limit = ConcurrencyLimit.of(ThreadSchedule("concurrency(10) rate(5/sec) random_arrivals(1 min)")) + assertFalse(limit.isUnlimited, "isUnlimited") + assertEquals(10, limit.limitAt(0.0)) + assertEquals(10, limit.limitAt(30.0)) + assertEquals(10, limit.limitAt(120.0)) + } + + @Test + fun `concurrency changes on the arrival boundary regardless of the generator position`() { + // 60 seconds at limit 10, then 120 seconds at limit 20 + val limit = ConcurrencyLimit.of( + ThreadSchedule( + "concurrency(10) rate(5/sec) random_arrivals(1 min) concurrency(20) random_arrivals(2 min)" + ) + ) + assertEquals(10, limit.limitAt(0.0), "at 0s") + assertEquals(10, limit.limitAt(59.9), "just before the boundary") + assertEquals(20, limit.limitAt(60.0), "at the boundary") + assertEquals(20, limit.limitAt(200.0), "after the boundary") + assertEquals(60.0, limit.nextChangeAfter(0.0), "next change from the start") + assertEquals(Double.POSITIVE_INFINITY, limit.nextChangeAfter(60.0), "no further changes") + } + + @Test + fun `last concurrency step at the same time wins`() { + val limit = ConcurrencyLimit.of(ThreadSchedule("concurrency(10) concurrency(20) rate(1/sec) random_arrivals(1 min)")) + assertEquals(20, limit.limitAt(0.0)) + } +} diff --git a/src/core/src/test/kotlin/org/apache/jmeter/threads/openmodel/ConcurrencyParseErrorTest.kt b/src/core/src/test/kotlin/org/apache/jmeter/threads/openmodel/ConcurrencyParseErrorTest.kt new file mode 100644 index 00000000000..b8e406ecc26 --- /dev/null +++ b/src/core/src/test/kotlin/org/apache/jmeter/threads/openmodel/ConcurrencyParseErrorTest.kt @@ -0,0 +1,40 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.jmeter.threads.openmodel + +import org.junit.jupiter.api.Assertions.assertThrows +import org.junit.jupiter.params.ParameterizedTest +import org.junit.jupiter.params.provider.ValueSource + +class ConcurrencyParseErrorTest { + @ParameterizedTest + @ValueSource( + strings = [ + "concurrency(0) rate(1/sec) random_arrivals(1 min)", + "concurrency(-1) rate(1/sec) random_arrivals(1 min)", + "concurrency(1.5) rate(1/sec) random_arrivals(1 min)", + "concurrency() rate(1/sec) random_arrivals(1 min)", + "concurrency(10 rate(1/sec) random_arrivals(1 min)", + ] + ) + fun `invalid concurrency is rejected`(schedule: String) { + assertThrows(Exception::class.java) { + ThreadSchedule(schedule) + } + } +} diff --git a/src/core/src/test/kotlin/org/apache/jmeter/threads/openmodel/ThreadScheduleTest.kt b/src/core/src/test/kotlin/org/apache/jmeter/threads/openmodel/ThreadScheduleTest.kt index fc822109463..12b1821b067 100644 --- a/src/core/src/test/kotlin/org/apache/jmeter/threads/openmodel/ThreadScheduleTest.kt +++ b/src/core/src/test/kotlin/org/apache/jmeter/threads/openmodel/ThreadScheduleTest.kt @@ -46,6 +46,15 @@ class ThreadScheduleTest { Case( "rate(0 per min) even_arrivals(30 min) rate(50 per min) random_arrivals(4 min) rate(200 per min) random_arrivals(10 sec)", "[Rate(0), Arrivals(type=EVEN, duration=1800), Rate(0.8), Arrivals(type=RANDOM, duration=240), Rate(3.3), Arrivals(type=RANDOM, duration=10)]" + ), + Case("concurrency(10)", "[Concurrency(10)]"), + Case( + "concurrency(10) rate(5/sec) random_arrivals(1 min)", + "[Concurrency(10), Rate(5), Arrivals(type=RANDOM, duration=60)]" + ), + Case( + "CONCURRENCY(10) rate(5/sec) random_arrivals(1 min) concurrency(20) random_arrivals(2 min)", + "[Concurrency(10), Rate(5), Arrivals(type=RANDOM, duration=60), Concurrency(20), Arrivals(type=RANDOM, duration=120)]" ) ) } diff --git a/src/core/src/test/kotlin/org/apache/jmeter/threads/openmodel/gui/TargetRateChartTest.kt b/src/core/src/test/kotlin/org/apache/jmeter/threads/openmodel/gui/TargetRateChartTest.kt new file mode 100644 index 00000000000..18ca7818b44 --- /dev/null +++ b/src/core/src/test/kotlin/org/apache/jmeter/threads/openmodel/gui/TargetRateChartTest.kt @@ -0,0 +1,66 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.jmeter.threads.openmodel.gui + +import org.apache.jmeter.threads.openmodel.ThreadSchedule +import org.jetbrains.letsPlot.core.util.MonolithicCommon +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test + +class TargetRateChartTest { + private val chart = TargetRateChart() + + private fun specFor(schedule: String) = chart.buildSpec(ThreadSchedule(schedule)) + + private fun assertLetsPlotAccepts(spec: MutableMap) { + val processed = MonolithicCommon.processRawSpecs(spec, frontendOnly = false) + assertFalse( + processed["kind"] == "error" || processed.containsKey("errorMessage"), + "lets-plot should process the spec without an error, but got: $processed" + ) + } + + @Test + fun `rate only schedule renders a single plot`() { + val spec = specFor("rate(10/sec) random_arrivals(1 min)") + assertEquals("plot", spec["kind"], "rate-only schedule should not add a concurrency panel") + assertLetsPlotAccepts(spec) + } + + @Test + fun `constant concurrency renders a single plot with a subtitle`() { + val spec = specFor("concurrency(10) rate(10/sec) random_arrivals(1 min)") + assertEquals("plot", spec["kind"], "a constant limit should be a subtitle, not a panel") + assertTrue( + spec.toString().contains("Max threads: 10"), + "the constant limit should appear as a subtitle, but got: $spec" + ) + assertLetsPlotAccepts(spec) + } + + @Test + fun `variable concurrency renders a stacked second panel`() { + val spec = specFor( + "concurrency(10) rate(10/sec) random_arrivals(1 min) concurrency(20) random_arrivals(1 min)" + ) + assertEquals("subplots", spec["kind"], "a changing limit should be drawn as a stacked panel") + assertLetsPlotAccepts(spec) + } +} diff --git a/xdocs/changes.xml b/xdocs/changes.xml index 988377c81b1..52078614442 100644 --- a/xdocs/changes.xml +++ b/xdocs/changes.xml @@ -69,6 +69,9 @@ Summary
  • 6274 Change references to old MySQL driver to new class com.mysql.cj.jdbc.Driver
  • 6352 Calculate delays in Open Model Thread Group and Precise Throughput Timer relative to start of Thread Group instead of the start of the test.
    • +
  • 5966 Open Model Thread Group: add a concurrency(...) schedule step that caps the number + of threads running at the same time, and an openmodel.max_concurrency property as an installation-wide + safety net against running out of memory. The schedule preview plots a variable limit as a second panel below the rate.
  • 63576358 Ensure writable directories when copying template files while report generation.
  • 65096675 Synchronize recent file menu across multiple JVMs. Contributed by Corneliu C (https://github.com/KingRabbid)
  • 6596Fallback to English locale when loading test plans that use string values for enum properties, so old sample plans load correctly even with non-English locales.
    • diff --git a/xdocs/usermanual/component_reference.xml b/xdocs/usermanual/component_reference.xml index 663b24fe6e4..e6e0997b311 100644 --- a/xdocs/usermanual/component_reference.xml +++ b/xdocs/usermanual/component_reference.xml @@ -6435,6 +6435,10 @@ See JMeter's Classpath and
    rate(10) random_arrivals(1 min) pause(2 sec) random_arrivals(1 min)
    run with constant load of ten requests per second during one minute, then make two second pause, then resume the load of ten requests per second for one minute
    + +
    concurrency(100) rate(10/sec) random_arrivals(1 min)
    +
    ten requests per second for one minute, with at most one hundred threads running at the same time. + If the server is slow and the arrivals would need more than one hundred threads, the extra arrivals wait for a free thread.

    The following commands are available: @@ -6445,6 +6449,19 @@ See JMeter's Classpath and You can omit time unit in case the rate is 0: rate(0) +

    concurrency(<number>)
    +
    limits the number of threads that run at the same time to the given positive integer. + The limit applies to the arrivals that follow the step, so an arrival that would exceed the limit waits for a + running thread to finish instead of starting a new one. Like rate, the limit is positional, so you can raise or lower it + during the test: concurrency(100) rate(10/sec) random_arrivals(10 min) concurrency(200) random_arrivals(10 min) + caps the load at one hundred threads for the first ten minutes and two hundred for the next ten. +
    The limit serves two purposes: it protects JMeter from running out of memory when the server responds slower than + arrivals are scheduled, and it models a system with a hard concurrency limit, such as a connection pool of a fixed size. +
    When the limit is saturated, JMeter no longer follows the open model: the achieved rate drops below the target rate, + which the reports show as fewer samples than requested. Arrivals that still have no free thread when the schedule ends are dropped. + To reach the target rate, raise the limit or reduce the rate. +
    +
    random_arrivals(<number> sec)
    configures random arrivals schedule with the given duration. The starting load rate is configured before random_arrivals, and the finish load rate is configured after random_arrivals. @@ -6485,6 +6502,10 @@ See JMeter's Classpath and If you want to let the threads complete safely, consider adding pause(5 min) at the end of the schedule. That will add five minutes for the threads to continue.

    +

    Set the openmodel.max_concurrency property to cap the number of threads for every Open Model Thread Group + that does not set its own concurrency(...). It is a safety net against running out of memory. The default is + unlimited, and a concurrency(...) step in the schedule overrides it. +

    There are no special functions for generating the load profile in a loop, however, the default JMeter templating functions can be helpful for generating the schedule.