Dynamic Load Adjustment in FunTester: A Real‑World Pressure‑Control Demo

Background and Goal

The FunTester framework already supports functional performance testing, but a dynamic pressure control feature is needed. The goal is to allow the test pressure (e.g., thread count or QPS) to be increased, decreased, or held during a single test run, enabling multiple scenarios without restarting the test.

Requirement Scenario

Typical use case: a system is expected to handle 10,000 QPS with 100 threads. The test starts from 0 threads, adds 10 threads every 30 seconds up to a maximum of 120, but may need to hold at 80 threads once a bottleneck is reached to collect data.

Design Overview

Basic Implementation

The solution reuses the soft‑start approach from a previous exploration, employing a for loop to ramp up pressure and a console‑based trigger to stop the ramp.

Trigger Condition

A thread‑safe variable com.funtester.config.Constant#INTPUT_KEY is monitored. When the user types the key, the variable com.funtester.frame.execute.HoldConcurrent#HOLD is set to 1, marking the test to pause further pressure increase.

Multithread Synchronization

Instead of CountDownLatch, the implementation uses java.util.concurrent.Phaser because it allows flexible registration and deregistration of threads during execution.

Open Issues

Local data statistics become inaccurate when the thread count changes dynamically.

Multi‑task support and service‑oriented deployment are not yet implemented.

Demo Implementation

Multithread Task Class

private static class FunTester implements Runnable {
    @Override
    public void run() {
        waitForKey(INTPUT_KEY);
        HOLD.set(1);
        output("压力暂停");
    }
}

public abstract class HoldThread<F> extends ThreadBase<F> {
    @Override
    public void before() {
        super.before();
        HoldConcurrent.phaser.register();
    }

    @Override
    protected void after() {
        super.after();
        GCThread.stop();
        HoldConcurrent.phaser.arriveAndDeregister();
    }
}

Execution Class

public class HoldConcurrent extends SourceCode {
    public static AtomicInteger HOLD = new AtomicInteger(0);
    private long startTime;
    private long endTime;
    public String desc;
    public List<ThreadBase> threads = new ArrayList<>();
    public int threadNum;
    public static Vector<Short> allTimes = new Vector<>();
    public static Vector<String> requestMark = new Vector<>();
    ExecutorService executorService;
    public static Phaser phaser;

    public HoldConcurrent(ThreadBase thread, int threadNum, String desc) {
        this(threadNum, desc);
        range(threadNum).forEach(x -> threads.add(thread.clone()));
    }

    private HoldConcurrent(int threadNum, String desc) {
        this.threadNum = threadNum;
        this.desc = StatisticsUtil.getFileName(desc);
        phaser = new Phaser(1);
        executorService = ThreadPoolUtil.createFixedPool(threadNum);
    }

    public PerformanceResultBean start() {
        Thread funtester = new Thread(new FunTester());
        funtester.start();
        ThreadBase.progress = new Progress(threads, StatisticsUtil.getTrueName(desc));
        ThreadBase.progress.threadNum = 0;
        new Thread(ThreadBase.progress).start();
        startTime = Time.getTimeStamp();
        for (int i = 0; i < threadNum; i++) {
            if (HOLD.get() == 1) { threadNum = i; break; }
            ThreadBase thread = threads.get(i);
            if (StringUtils.isBlank(thread.threadName))
                thread.threadName = StatisticsUtil.getTrueName(desc) + i;
            sleep(RUNUP_TIME / threadNum);
            executorService.execute(thread);
            ThreadBase.progress.threadNum = i + 1;
            logger.info("已经启动了 {} 个线程!", i + 1);
        }
        phaser.arriveAndAwaitAdvance();
        executorService.shutdown();
        ThreadBase.progress.stop();
        // collect statistics …
        endTime = Time.getTimeStamp();
        HOLD.set(0);
        logger.info("总计{}个线程，共用时：{} s,执行总数:{},错误数:{},失败数:{}",
            threadNum, Time.getTimeDiffer(startTime, endTime), formatLong(executeTotal), errorTotal, failTotal);
        return over();
    }
    // helper methods omitted for brevity
}

Test Script

package com.funtest.groovytest;

import com.funtester.base.constaint.HoldThread;
import com.funtester.base.constaint.ThreadBase;
import com.funtester.frame.SourceCode;
import com.funtester.frame.execute.HoldConcurrent;

/** * 加压暂停 */
class HoldTest extends SourceCode {
    public static void main(String[] args) {
        def tester = new FunTester();
        new HoldConcurrent(tester, 10, "终止压力递增保持压力测试").start();
    }

    private static class FunTester extends HoldThread {
        FunTester() { super(null, 100, true); }
        @Override protected void doing() throws Exception { sleep(0.1); }
        @Override ThreadBase clone() { return new FunTester(); }
    }
}

Console Output

INFO-> 当前用户：oker，工作目录：/Users/oker/IdeaProjects/funtester/src/test/groovy/com/funtest/groovytest/,系统编码格式:UTF-8,系统Mac OS X版本:10.16
WARN-> 请输入“FunTester”继续运行！
INFO-> 已经启动了 1 个线程!
INFO-> 终止压力递增保持压力测试进度:▍  2% ,当前QPS: 0
INFO-> 已经启动了 2 个线程!
INFO-> 终止压力递增保持压力测试进度:▍▍▍▍▍▍▍▍▍▍▍▍▍▍▍▍▍▍▍▍  31% ,当前QPS: 20
... (output continues) ...
INFO-> 总计4个线程，共用时：13.439 s,执行总数:227,错误数:0,失败数:0
INFO-> 数据保存成功！文件名：/Users/oker/IdeaProjects/funtester/src/test/groovy/com/funtest/groovytest/long/data/终止压力递增保持压力测试231809_4

Observations

The demo confirms that pressure can be paused via console input. The reported QPS and QPS2 differ by more than a factor of two because they are calculated using different formulas. The current implementation lacks a resume capability and does not support multiple concurrent tasks.

Future Work

Planned improvements include adding a resume function after pause, supporting multiple concurrent tasks, and extending the model to a service‑oriented architecture for broader performance‑testing scenarios.

References

Performance Test Soft‑Start Exploration; Java Thread Synchronization Three‑Sword.