feat: add concurrency limit to Open Model Thread Group (#5966)

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
 #---------------------------------------------------------------------------

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
+            )
+        }
+    }
+}

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<Change> 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<Double>()
+            val values = mutableListOf<Int>()
+            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())
+        }
+    }
+}