This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12896-ee83c538ebfadd29d60d5622afd9a12213f02d36 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 99ea16b9d6bdce452d4c4955581ccc7236a81bbd Author: Gangavarapu Vivek <[email protected]> AuthorDate: Thu Oct 22 05:07:34 2026 +0000 [Test][Zeta] Warm up the checkpoint scheduling benchmark by trigger count (#12896) --- .../benchmark/CheckpointSchedulingBenchmark.java | 44 +++++++++++++-- .../checkpoint/CheckpointSchedulingFixture.java | 64 +++++++++++++++++++++- .../checkpoint/FakeTaskCheckpointManager.java | 8 ++- .../CheckpointSchedulingBenchmarkTest.java | 34 ++++++++++++ .../CheckpointSchedulingFixtureTest.java | 42 ++++++++++++++ 5 files changed, 182 insertions(+), 10 deletions(-) diff --git a/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/CheckpointSchedulingBenchmark.java b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/CheckpointSchedulingBenchmark.java index 8e2b855310..7e11a3a18a 100644 --- a/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/CheckpointSchedulingBenchmark.java +++ b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/CheckpointSchedulingBenchmark.java @@ -59,10 +59,25 @@ import java.util.concurrent.TimeUnit; * * <p>The checkpoint interval is {@code pipelineNum * triggerSpacingMillis}, so every point of the * sweep sees the same rate of due triggers and the same checkpoint load on storage, and only the - * number of coordinators changes. It is floored at {@link #MIN_MEASURABLE_INTERVAL_MILLIS}, so the - * smallest pipeline counts sample less often. Each job has one pipeline: many jobs is what "many - * active pipelines" means on a member, and it keeps per-job checkpoint state from becoming the - * bottleneck. + * number of coordinators changes. It is floored at {@link #MIN_MEASURABLE_INTERVAL_MILLIS}, which + * is why the sweep starts at ten pipelines: below that the floor binds, the member triggers fewer + * checkpoints a second than at the other points, and the code on the trigger path warms up that + * much slower. Each job has one pipeline: many jobs is what "many active pipelines" means on a + * member, and it keeps per-job checkpoint state from becoming the bottleneck. + * + * <p>The trigger path runs only 50 times a second, and the JIT compiles by invocation count, so + * with default thresholds that path is still being compiled minutes into a fork, reaching C2 only + * after about two minutes. Every iteration in that window scores lower than the one before. Two + * things keep that out of the measurement. The forks run with {@code + * -XX:CompileThresholdScaling=0.05}, so the path compiles after a twentieth of the invocations; + * lower values compile so much at once that triggers come due unmeasured and the skip limit fails + * the run. And each trial first lets its coordinators trigger {@link #TRIGGER_PATH_WARMUP_TRIGGERS} + * checkpoints, a warmup counted in triggers rather than seconds, and runs for at least {@link + * #MIN_WARMUP_SECONDS}. At 50 triggers a second the time is what binds; the count keeps the warmup + * long enough if a lower trigger rate is added. + * + * <p>Scores taken before this warmup are not comparable with scores taken after it, so a PR + * comparison needs the candidate branch to contain it as well as the baseline. */ @BenchmarkMode(Mode.SampleTime) @OutputTimeUnit(TimeUnit.MICROSECONDS) @@ -78,6 +93,7 @@ import java.util.concurrent.TimeUnit; "-XX:+AlwaysPreTouch", "-XX:+DisableExplicitGC", "-XX:ActiveProcessorCount=4", + "-XX:CompileThresholdScaling=0.05", "-Djava.net.preferIPv4Stack=true" }) public class CheckpointSchedulingBenchmark extends BenchmarkBase { @@ -90,6 +106,20 @@ public class CheckpointSchedulingBenchmark extends BenchmarkBase { */ static final long MIN_MEASURABLE_INTERVAL_MILLIS = 200L; + /** + * Checkpoints each trial triggers before JMH's own warmup starts. With the fork's lowered + * compile thresholds, the trigger path, the scheduler queue and the MDC wrapper around each + * scheduled task stopped changing tier after about 1,000 triggers on GitHub's Linux runners. + */ + static final long TRIGGER_PATH_WARMUP_TRIGGERS = 1_000L; + + /** + * Least time each trial runs its coordinators before JMH's own warmup. At ten pipelines or more + * the trigger count above is reached in 20 seconds, while the rest of the member is still + * compiling, and the first measured iteration then caught millisecond outliers. + */ + static final long MIN_WARMUP_SECONDS = 60L; + public static void main(String[] args) throws RunnerException { Options options = new OptionsBuilder() @@ -111,7 +141,7 @@ public class CheckpointSchedulingBenchmark extends BenchmarkBase { @State(Scope.Thread) public static class CoordinatorsState { - @Param({"1", "10", "100", "500"}) + @Param({"10", "100", "500"}) private int pipelineNum; @Param({"20"}) @@ -126,7 +156,9 @@ public class CheckpointSchedulingBenchmark extends BenchmarkBase { pipelineNum, Math.max( MIN_MEASURABLE_INTERVAL_MILLIS, - pipelineNum * triggerSpacingMillis)); + pipelineNum * triggerSpacingMillis), + TRIGGER_PATH_WARMUP_TRIGGERS, + TimeUnit.SECONDS.toMillis(MIN_WARMUP_SECONDS)); fixture.setUp(); System.out.printf( "# checkpoint scheduler threads for %d pipelines: %d; measuring %d of them%n", diff --git a/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/checkpoint/CheckpointSchedulingFixture.java b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/checkpoint/CheckpointSchedulingFixture.java index 2dc7cccb54..0ebe7b0153 100644 --- a/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/checkpoint/CheckpointSchedulingFixture.java +++ b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/checkpoint/CheckpointSchedulingFixture.java @@ -41,6 +41,7 @@ import java.util.concurrent.SynchronousQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.LockSupport; import java.util.stream.Collectors; @@ -127,9 +128,14 @@ public final class CheckpointSchedulingFixture { private static final long EXECUTOR_KEEP_ALIVE_SECONDS = 60L; private static final long SHUTDOWN_TIMEOUT_SECONDS = 30L; + private static final long WARMUP_POLL_NANOS = TimeUnit.MILLISECONDS.toNanos(10); + private final int pipelineNum; private final long intervalMillis; private final long intervalNanos; + private final long warmupTriggers; + private final long minWarmupNanos; + private final AtomicLong triggeredCheckpoints = new AtomicLong(); private int probeStride; private Set<Thread> preexistingSchedulerThreads = Collections.emptySet(); @@ -154,15 +160,29 @@ public final class CheckpointSchedulingFixture { * @param intervalMillis checkpoint interval of every job */ public CheckpointSchedulingFixture(int pipelineNum, long intervalMillis) { + this(pipelineNum, intervalMillis, 0L, 0L); + } + + /** + * @param pipelineNum number of jobs, each with one single-task pipeline + * @param intervalMillis checkpoint interval of every job + * @param warmupTriggers checkpoints the coordinators trigger before setup returns, 0 for none + * @param minWarmupMillis least time the coordinators run before setup returns, 0 for none + */ + public CheckpointSchedulingFixture( + int pipelineNum, long intervalMillis, long warmupTriggers, long minWarmupMillis) { this.pipelineNum = pipelineNum; this.intervalMillis = intervalMillis; this.intervalNanos = TimeUnit.MILLISECONDS.toNanos(intervalMillis); + this.warmupTriggers = warmupTriggers; + this.minWarmupNanos = TimeUnit.MILLISECONDS.toNanos(minWarmupMillis); } /** * Starts the member, creates the coordinators and starts their pipelines staggered across one - * interval, so due triggers arrive as a steady stream rather than a burst. Returns once every - * coordinator's trigger phase is known. + * interval, so due triggers arrive as a steady stream rather than a burst. Then lets them + * trigger {@code warmupTriggers} checkpoints for at least {@code minWarmupMillis}, and returns + * once every coordinator's trigger phase is known. */ public void setUp() throws Exception { validateParameters(); @@ -175,11 +195,44 @@ public final class CheckpointSchedulingFixture { coordinatorExecutor = createCoordinatorExecutor(engineConfig); createManagers(server, engineConfig.getCheckpointConfig()); startStaggered(); + awaitWarmupTriggers(); // Every coordinator, not only the probes: once each has triggered, each has armed its // scheduler, so countSchedulerThreads() sees the full thread cost of pipelineNum pipelines. resync(1); } + /** + * Waits until the coordinators have triggered {@code warmupTriggers} checkpoints. The JIT + * compiles the trigger path by how often it runs, not by how long the fork has been running, so + * a warmup counted in triggers gives every pipeline count the same compiled code to measure, + * where a warmup counted in seconds gives the lowest trigger rates the least. Nothing is + * measured meanwhile, and the coordinators are the measured ones, so no code path is exercised + * that the measurement does not also take. The minimum time covers the rest of the member, + * whose compilation goes on for a while after the trigger path settles at the higher rates. + */ + private void awaitWarmupTriggers() { + long start = System.nanoTime(); + long target = triggeredCheckpoints.get() + warmupTriggers; + long earliestEnd = start + minWarmupNanos; + // Twice the time the triggers take at the configured rate, plus the pending re-arms. + long deadline = + Math.max(earliestEnd, start + 2 * warmupTriggers * intervalNanos / pipelineNum) + + RESYNC_INTERVALS * intervalNanos + + PENDING_REARM_NANOS; + while (triggeredCheckpoints.get() < target || System.nanoTime() < earliestEnd) { + checkFailure(); + if (System.nanoTime() > deadline) { + throw new IllegalStateException( + String.format( + "%d pipelines triggered %d of %d warm-up checkpoints in time", + pipelineNum, + warmupTriggers - (target - triggeredCheckpoints.get()), + warmupTriggers)); + } + parkUntil(System.nanoTime() + WARMUP_POLL_NANOS); + } + } + /** Resynchronises the coordinators that were taken out of the rotation, then resets counts. */ public void beginIteration() { resync(probeStride); @@ -345,6 +398,10 @@ public final class CheckpointSchedulingFixture { return sampled; } + long getTriggeredCheckpoints() { + return triggeredCheckpoints.get(); + } + /** * Counts this fixture's live checkpoint scheduler threads. This is the cost a shared scheduler * exists to remove, so it is reported rather than derived. @@ -393,7 +450,8 @@ public final class CheckpointSchedulingFixture { runningJobState, server.getEngineContext(), server.getCheckpointMonitorService(), - failure); + failure, + triggeredCheckpoints); managers.add(manager); pendingCounters[i] = readPendingCounter(manager.getCheckpointCoordinator(PIPELINE_ID)); lastTriggerNanos[i] = NOT_SYNCED; diff --git a/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/checkpoint/FakeTaskCheckpointManager.java b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/checkpoint/FakeTaskCheckpointManager.java index d57bed4a45..d3942b2b43 100644 --- a/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/checkpoint/FakeTaskCheckpointManager.java +++ b/seatunnel-benchmarks/src/main/java/org/apache/seatunnel/benchmark/checkpoint/FakeTaskCheckpointManager.java @@ -40,6 +40,7 @@ import com.hazelcast.spi.impl.operationservice.impl.InvocationFuture; import java.lang.reflect.Field; import java.util.Collections; import java.util.concurrent.ExecutorService; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; /** @@ -76,7 +77,9 @@ final class FakeTaskCheckpointManager extends CheckpointManager { private final NodeEngine nodeEngine; private final CheckpointPlan plan; private final AtomicReference<Throwable> failure; + private final AtomicLong triggeredCheckpoints; + /** @param triggeredCheckpoints incremented for every checkpoint barrier a trigger sends */ FakeTaskCheckpointManager( long jobId, NodeEngine nodeEngine, @@ -87,7 +90,8 @@ final class FakeTaskCheckpointManager extends CheckpointManager { IMap<Object, Object> runningJobStateIMap, SeaTunnelEngineContext engineContext, CheckpointMonitorService checkpointMonitorService, - AtomicReference<Throwable> failure) { + AtomicReference<Throwable> failure, + AtomicLong triggeredCheckpoints) { super( jobId, false, @@ -106,6 +110,7 @@ final class FakeTaskCheckpointManager extends CheckpointManager { this.nodeEngine = nodeEngine; this.plan = plan; this.failure = failure; + this.triggeredCheckpoints = triggeredCheckpoints; } /** @@ -123,6 +128,7 @@ final class FakeTaskCheckpointManager extends CheckpointManager { @Override protected InvocationFuture<?> sendOperationToMemberNode(TaskOperation operation) { if (operation instanceof CheckpointBarrierTriggerOperation) { + triggeredCheckpoints.incrementAndGet(); acknowledgeBarrier(readBarrier((CheckpointBarrierTriggerOperation) operation)); } return NodeEngineUtil.sendOperationToMemberNode( diff --git a/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/CheckpointSchedulingBenchmarkTest.java b/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/CheckpointSchedulingBenchmarkTest.java index a9217bde95..01d80dfa85 100644 --- a/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/CheckpointSchedulingBenchmarkTest.java +++ b/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/CheckpointSchedulingBenchmarkTest.java @@ -19,13 +19,17 @@ package org.apache.seatunnel.benchmark; import org.junit.jupiter.api.Test; import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; import org.openjdk.jmh.annotations.Mode; import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Param; import org.openjdk.jmh.annotations.Threads; +import java.util.Arrays; import java.util.concurrent.TimeUnit; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; class CheckpointSchedulingBenchmarkTest { @@ -39,4 +43,34 @@ class CheckpointSchedulingBenchmarkTest { CheckpointSchedulingBenchmark.class.getAnnotation(OutputTimeUnit.class).value()); assertEquals(1, CheckpointSchedulingBenchmark.class.getAnnotation(Threads.class).value()); } + + @Test + void shouldLowerJitThresholdsSoTheTriggerPathCompilesWithinWarmup() { + assertTrue( + Arrays.asList( + CheckpointSchedulingBenchmark.class + .getAnnotation(Fork.class) + .jvmArgsAppend()) + .contains("-XX:CompileThresholdScaling=0.05")); + } + + @Test + void shouldOnlySweepPipelineCountsTheIntervalFloorDoesNotSlowDown() throws Exception { + String[] pipelineNums = paramValues("pipelineNum"); + long triggerSpacingMillis = Long.parseLong(paramValues("triggerSpacingMillis")[0]); + + for (String pipelineNum : pipelineNums) { + assertTrue( + Integer.parseInt(pipelineNum) * triggerSpacingMillis + >= CheckpointSchedulingBenchmark.MIN_MEASURABLE_INTERVAL_MILLIS, + "pipelineNum=" + pipelineNum + " would trigger below the sweep's rate"); + } + } + + private static String[] paramValues(String field) throws NoSuchFieldException { + return CheckpointSchedulingBenchmark.CoordinatorsState.class + .getDeclaredField(field) + .getAnnotation(Param.class) + .value(); + } } diff --git a/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/checkpoint/CheckpointSchedulingFixtureTest.java b/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/checkpoint/CheckpointSchedulingFixtureTest.java index c28b4786b2..36bbd517af 100644 --- a/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/checkpoint/CheckpointSchedulingFixtureTest.java +++ b/seatunnel-benchmarks/src/test/java/org/apache/seatunnel/benchmark/checkpoint/CheckpointSchedulingFixtureTest.java @@ -34,6 +34,8 @@ class CheckpointSchedulingFixtureTest { private static final long CHECKPOINT_INTERVAL_MILLIS = 200L; private static final int SAMPLE_COUNT = 20; private static final long THREAD_STOP_TIMEOUT_NANOS = TimeUnit.SECONDS.toNanos(30); + private static final long WARMUP_TRIGGERS = 40L; + private static final long MIN_WARMUP_MILLIS = 5_000L; @Test void shouldMeasureDueTriggersOfRealCoordinators() throws Exception { @@ -61,6 +63,46 @@ class CheckpointSchedulingFixtureTest { } } + @Test + void shouldTriggerTheWarmupCheckpointsBeforeMeasuringAnything() throws Exception { + CheckpointSchedulingFixture fixture = + new CheckpointSchedulingFixture( + PIPELINE_NUM, CHECKPOINT_INTERVAL_MILLIS, WARMUP_TRIGGERS, 0L); + fixture.setUp(); + try { + assertTrue( + fixture.getTriggeredCheckpoints() >= WARMUP_TRIGGERS, + "setup should return only after the warm-up checkpoints were triggered"); + // Phases are taken after the warm-up, so the first iteration skips nothing for them. + fixture.beginIteration(); + for (int i = 0; i < SAMPLE_COUNT; i++) { + fixture.awaitNextDueTrigger(); + fixture.awaitTrigger(); + } + assertEquals(SAMPLE_COUNT, fixture.getSampled()); + fixture.endIteration(); + } finally { + fixture.tearDown(); + } + } + + @Test + void shouldKeepWarmingUpForTheMinimumTimeOnceTheTriggersRan() throws Exception { + // One trigger is done within a second; the minimum time is what setup waits for. + CheckpointSchedulingFixture fixture = + new CheckpointSchedulingFixture( + PIPELINE_NUM, CHECKPOINT_INTERVAL_MILLIS, 1L, MIN_WARMUP_MILLIS); + long start = System.nanoTime(); + fixture.setUp(); + try { + assertTrue( + System.nanoTime() - start >= TimeUnit.MILLISECONDS.toNanos(MIN_WARMUP_MILLIS), + "setup should not return before the minimum warm-up time"); + } finally { + fixture.tearDown(); + } + } + @Test void shouldStopEverySchedulerThreadOnTearDown() throws Exception { // Stands in for a scheduler thread another test in this JVM has not finished stopping.
