kevin-wu24 commented on code in PR #22669: URL: https://github.com/apache/kafka/pull/22669#discussion_r3484851467
########## jmh-benchmarks/src/main/java/org/apache/kafka/jmh/raft/ElectionBenchmarks.java: ########## @@ -0,0 +1,78 @@ +/* + * 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.kafka.jmh.raft; + +import org.apache.kafka.raft.RaftClientBenchmarkContext; +import org.apache.kafka.raft.RaftClientTestContext; + +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.Warmup; + +import java.util.Optional; +import java.util.concurrent.TimeUnit; + +/** + * Benchmarks for the leader-election path. The outer class is intentionally not a JMH {@code @State}: + * each benchmark declares the starting state it needs as a nested {@code @State} parameter, so + * different election scenarios (e.g. a future Prospective or Candidate start) can have their own + * setup without forcing a single shared {@code @Setup} on the whole class. + */ +@BenchmarkMode(Mode.SingleShotTime) +@OutputTimeUnit(TimeUnit.NANOSECONDS) +@Warmup(iterations = RaftClientBenchmarkContext.SINGLE_SHOT_WARMUP_ITERATIONS) +@Measurement(iterations = RaftClientBenchmarkContext.SINGLE_SHOT_MEASUREMENT_ITERATIONS) +@Fork(RaftClientBenchmarkContext.SINGLE_SHOT_FORKS) +public class ElectionBenchmarks { + + /** + * Starting state: the local node is Unattached in a {@code voterCount}-node quorum. A fresh + * context is built per invocation because driving the election to completion consumes it. + */ + @State(Scope.Thread) + public static class UnattachedQuorum { Review Comment: A better name for this would be `UnattachedWithMultipleVoters`. The word `Unattached` means "I don't know the leader" for an epoch, whereas the word `Quorum` implies voters agreed on who a leader is for an epoch. ########## raft/src/testFixtures/java/org/apache/kafka/raft/RaftClientBenchmarkContext.java: ########## @@ -17,32 +17,51 @@ package org.apache.kafka.raft; import org.apache.kafka.common.Uuid; -import org.apache.kafka.common.message.FetchRequestData; -import org.apache.kafka.common.protocol.ApiMessage; +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.raft.RaftClientTestContext.RaftProtocol; import org.apache.kafka.server.common.KRaftVersion; +import java.util.HashMap; import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ThreadLocalRandom; import java.util.stream.Collectors; import java.util.stream.IntStream; public final class RaftClientBenchmarkContext { + // Standardized JMH iteration counts, shared by all raft benchmarks so every benchmark of a given + // mode is configured identically. SingleShotTime measures a single operation per iteration, so it + // needs many iterations to build a stable distribution; AverageTime averages many operations + // within each timed iteration, so it needs fewer. + public static final int SINGLE_SHOT_WARMUP_ITERATIONS = 50; + public static final int SINGLE_SHOT_MEASUREMENT_ITERATIONS = 30; + public static final int SINGLE_SHOT_FORKS = 5; + public static final int AVERAGE_TIME_WARMUP_ITERATIONS = 5; + public static final int AVERAGE_TIME_MEASUREMENT_ITERATIONS = 10; + public static final int AVERAGE_TIME_FORKS = 3; + + public static final KRaftVersion DEFAULT_KRAFT_VERSION = KRaftVersion.KRAFT_VERSION_1; + public static final RaftProtocol DEFAULT_RAFT_PROTOCOL = RaftProtocol.KIP_1186_PROTOCOL; Review Comment: We should grab the most highest ordinal value of these enums so we do not need to keep updating this. ``` LATEST_KRAFT_VERSION = Arrays.stream(KRaftVersion.values()) .max(Comparator.naturalOrder()) .orElseThrow() ``` ########## raft/src/testFixtures/java/org/apache/kafka/raft/RaftClientBenchmarkContext.java: ########## @@ -17,32 +17,51 @@ package org.apache.kafka.raft; import org.apache.kafka.common.Uuid; -import org.apache.kafka.common.message.FetchRequestData; -import org.apache.kafka.common.protocol.ApiMessage; +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.raft.RaftClientTestContext.RaftProtocol; import org.apache.kafka.server.common.KRaftVersion; +import java.util.HashMap; import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ThreadLocalRandom; import java.util.stream.Collectors; import java.util.stream.IntStream; public final class RaftClientBenchmarkContext { + // Standardized JMH iteration counts, shared by all raft benchmarks so every benchmark of a given + // mode is configured identically. SingleShotTime measures a single operation per iteration, so it + // needs many iterations to build a stable distribution; AverageTime averages many operations + // within each timed iteration, so it needs fewer. + public static final int SINGLE_SHOT_WARMUP_ITERATIONS = 50; + public static final int SINGLE_SHOT_MEASUREMENT_ITERATIONS = 30; + public static final int SINGLE_SHOT_FORKS = 5; + public static final int AVERAGE_TIME_WARMUP_ITERATIONS = 5; + public static final int AVERAGE_TIME_MEASUREMENT_ITERATIONS = 10; + public static final int AVERAGE_TIME_FORKS = 3; Review Comment: You may want to break these comments up, so that the `SingleShotTime` documentation goes with the `SINGLE_SHOT_...` static variables, etc.. ########## raft/src/testFixtures/java/org/apache/kafka/raft/RaftClientBenchmarkContext.java: ########## @@ -51,93 +70,106 @@ private RaftClientBenchmarkContext(RaftClientTestContext context, ReplicaKey oth * Unattached → Leader election. A single-voter quorum is rejected because such a node elects * itself at initialization, before any measured poll. */ - public static RaftClientBenchmarkContext unattached(int localId, int voterCount) throws Exception { + public static RaftClientBenchmarkContext unattached(int voterCount) throws Exception { + return unattached(voterCount, DEFAULT_KRAFT_VERSION, DEFAULT_RAFT_PROTOCOL); + } + + public static RaftClientBenchmarkContext unattached( + int voterCount, + KRaftVersion kraftVersion, + RaftProtocol raftProtocol + ) throws Exception { if (voterCount < 2) { throw new IllegalArgumentException("voterCount must be at least 2; a single voter self-elects at init"); } - return new RaftClientBenchmarkContext(buildContext(voterKeys(localId, voterCount)), null); + List<ReplicaKey> voterKeys = voterKeys(voterCount); + return new RaftClientBenchmarkContext(buildContext(voterKeys, kraftVersion, raftProtocol), voterKeys); + } + + public static RaftClientBenchmarkContext leader(int voterCount) throws Exception { + return leader(voterCount, DEFAULT_KRAFT_VERSION, DEFAULT_RAFT_PROTOCOL); } - public static RaftClientBenchmarkContext leader(int localId, int voterCount) throws Exception { - List<ReplicaKey> voterKeys = voterKeys(localId, voterCount); - RaftClientTestContext context = buildContext(voterKeys); + public static RaftClientBenchmarkContext leader( + int voterCount, + KRaftVersion kraftVersion, + RaftProtocol raftProtocol + ) throws Exception { + List<ReplicaKey> voterKeys = voterKeys(voterCount); + RaftClientTestContext context = buildContext(voterKeys, kraftVersion, raftProtocol); context.unattachedToLeader(); - return new RaftClientBenchmarkContext(context, voterKeys.get(1)); + return new RaftClientBenchmarkContext(context, voterKeys); } - /** {@code voterCount} voter keys, each with a random directory id, with {@code localId} first. */ - private static List<ReplicaKey> voterKeys(int localId, int voterCount) { + /** + * {@code voterCount} voter keys, each with a random directory id, with the local node first. + */ + private static List<ReplicaKey> voterKeys(int voterCount) { + int localId = randomReplicaId(); return IntStream.range(0, voterCount) .mapToObj(i -> ReplicaKey.of(localId + i, Uuid.randomUuid())) .collect(Collectors.toList()); } + private static int randomReplicaId() { + return ThreadLocalRandom.current().nextInt(1025); + } + /** - * Builds an unattached node in a KIP-1186 quorum of {@code voterKeys}, whose first entry is the - * local node. + * Builds an unattached node in a quorum of {@code voterKeys} (first entry is the local node) */ - private static RaftClientTestContext buildContext(List<ReplicaKey> voterKeys) throws Exception { + private static RaftClientTestContext buildContext( + List<ReplicaKey> voterKeys, + KRaftVersion kraftVersion, + RaftProtocol raftProtocol + ) throws Exception { ReplicaKey local = voterKeys.get(0); VoterSet voters = VoterSetTestUtil.voterSet(voterKeys.stream()); return new RaftClientTestContext.Builder(local.id(), local.directoryId().get()) - .withStartingVoters(voters, KRaftVersion.KRAFT_VERSION_1) - .withRaftProtocol(RaftClientTestContext.RaftProtocol.KIP_1186_PROTOCOL) + .withStartingVoters(voters, kraftVersion) + .withRaftProtocol(raftProtocol) .withPollIntervalMs(0) .withUnknownLeader(0) .build(); } - public void unattachedToLeader() throws Exception { - context.unattachedToLeader(); - } - - public void advanceLeaderHwmToLogEnd() throws InterruptedException { - context.advanceLocalLeaderHighWatermarkToLogEndOffset(); - } - - public void pollUntilResponse() throws InterruptedException { - context.pollUntilResponse(); - } - - public int currentEpoch() { - return context.currentEpoch(); + public RaftClientTestContext testContext() { + return context; } + /** The local node's log end offset. Kept here because the {@code log} field is package-private. */ public long logEndOffset() { return log.endOffset().offset(); } - /** A voter other than the local node, e.g. to deliver a FETCH from on the leader. */ - public ReplicaKey otherVoter() { - if (otherVoter == null) { - throw new IllegalStateException("This context was not built with an other-voter key"); - } - return otherVoter; - } - - public void deliverRequest(ApiMessage request) { - context.deliverRequest(request); + public ReplicaKey localVoter() { + return voters.get(0); } - public FetchRequestData fetchRequest( - int epoch, - ReplicaKey replicaKey, - long fetchOffset, - int lastFetchedEpoch, - int maxWaitMs - ) { - return context.fetchRequest(epoch, replicaKey, fetchOffset, lastFetchedEpoch, maxWaitMs); + /** + * The voters other than the local node, in voter-set order. May be empty (single-voter quorum). + * Use these as the source of delivered requests, e.g. a FETCH from a follower on the leader. + */ + public List<ReplicaKey> remoteVoters() { + return voters.subList(1, voters.size()); } - /** Re-baselines all counters to the current totals. Call at the end of benchmark setup. */ - public void resetCounters() { + /** + * Establishes the counter baseline so that work done before this point (building the context, + * driving an election in {@code leader()}, or anything else a benchmark does in its setup) is not + * attributed to the measured operation. Call this at the <b>end of benchmark setup</b>, just + * before the measured region begins. + */ + public void zeroCountersOnSetup() { lastFlushCount = log.flushCount(); lastReadCount = log.readCount(); lastTruncationCount = log.truncationCount(); lastRequestsSent = channel.requestsSent(); + lastRequestsSentByApiKey = new HashMap<>(channel.requestsSentByApiKey()); Review Comment: We should be able to remove this too. ########## raft/src/testFixtures/java/org/apache/kafka/raft/RaftClientTestContext.java: ########## @@ -658,7 +665,7 @@ void expectAndGrantVotes(int epoch) throws Exception { deliverResponse(request.correlationId(), request.destination(), voteResponse); } - client.poll(); + pollUntil(() -> client.quorum().isLeader()); Review Comment: Hmm, is this a result of you trying to benchmark with 5 nodes? I guess in that case we need to handle multiple `VoteResponses` to get elected. ########## raft/src/testFixtures/java/org/apache/kafka/raft/RaftClientBenchmarkContext.java: ########## @@ -17,32 +17,51 @@ package org.apache.kafka.raft; import org.apache.kafka.common.Uuid; -import org.apache.kafka.common.message.FetchRequestData; -import org.apache.kafka.common.protocol.ApiMessage; +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.raft.RaftClientTestContext.RaftProtocol; import org.apache.kafka.server.common.KRaftVersion; +import java.util.HashMap; import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ThreadLocalRandom; import java.util.stream.Collectors; import java.util.stream.IntStream; public final class RaftClientBenchmarkContext { + // Standardized JMH iteration counts, shared by all raft benchmarks so every benchmark of a given + // mode is configured identically. SingleShotTime measures a single operation per iteration, so it + // needs many iterations to build a stable distribution; AverageTime averages many operations + // within each timed iteration, so it needs fewer. + public static final int SINGLE_SHOT_WARMUP_ITERATIONS = 50; + public static final int SINGLE_SHOT_MEASUREMENT_ITERATIONS = 30; + public static final int SINGLE_SHOT_FORKS = 5; + public static final int AVERAGE_TIME_WARMUP_ITERATIONS = 5; + public static final int AVERAGE_TIME_MEASUREMENT_ITERATIONS = 10; + public static final int AVERAGE_TIME_FORKS = 3; + + public static final KRaftVersion DEFAULT_KRAFT_VERSION = KRaftVersion.KRAFT_VERSION_1; + public static final RaftProtocol DEFAULT_RAFT_PROTOCOL = RaftProtocol.KIP_1186_PROTOCOL; + private final RaftClientTestContext context; private final MockLog log; private final MockNetworkChannel channel; - private final ReplicaKey otherVoter; + private final List<ReplicaKey> voters; private int lastFlushCount; private int lastReadCount; private int lastTruncationCount; private int lastRequestsSent; + private Map<Short, Integer> lastRequestsSentByApiKey = new HashMap<>(); Review Comment: We should be able to remove this. ########## raft/src/testFixtures/java/org/apache/kafka/raft/RaftClientBenchmarkContext.java: ########## @@ -162,10 +194,21 @@ public int drainLogTruncations() { return delta; } - public int drainRpcRequestsSent() { - int current = channel.requestsSent(); - int delta = current - lastRequestsSent; - lastRequestsSent = current; + /** + * Number of requests sent since the last drain. If {@code apiKey} is present, only requests of + * that API key are counted; otherwise all requests are counted. + */ + public int drainRpcRequestsSent(Optional<ApiKeys> apiKey) { + if (apiKey.isEmpty()) { + int current = channel.requestsSent(); + int delta = current - lastRequestsSent; + lastRequestsSent = current; + return delta; + } + short id = apiKey.get().id; + int current = channel.requestsSent(apiKey.get()); + int delta = current - lastRequestsSentByApiKey.getOrDefault(id, 0); + lastRequestsSentByApiKey.put(id, current); Review Comment: Seems like we are missing a call to `MockNetworkChannel#drainSentRequests(Optional<ApiKeys> apiKeyFilter)` before entering this method, since that needs to happen before we collect the request deltas for the current invocation. We should call that method in `drainFrom`. Also, we should be able to assert the `sendQueue` is empty at the end of this method. Otherwise, that is a performance regression, because we're sending more requests than we think. This is so we can drain any remaining sent requests after the benchmarking code calls `poll()` for the last time in the current invocation. Only after this will our delta collection be accurate. This is similar to the current implementation of `drainRpcResponsesSent`. Additionally, this `lastRequestsSentByApiKey` logic from L208-211 is going to be incorrect. Basically, we're not going to count any other requests sent over the benchmarking invocation if we pass an `apiKey` to this method. Maybe my previous comments about this weren't very clear, and I think the naming convention of these methods contributed to it. I think this was the comment: https://github.com/apache/kafka/pull/22669#discussion_r3475499363? Since we are only reporting the `requestsSent/invocation` at the end, this `lastRequestsSentByApiKey` state is not necessary. ########## jmh-benchmarks/src/main/java/org/apache/kafka/jmh/raft/KRaftBenchmarkingCounters.java: ########## @@ -0,0 +1,151 @@ +/* + * 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.kafka.jmh.raft; + +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.raft.RaftClientBenchmarkContext; + +import org.openjdk.jmh.annotations.AuxCounters; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.infra.BenchmarkParams; + +import java.util.Optional; + +/** + * Secondary, machine-independent work counters reported by the raft benchmarks alongside the timing + * score, as {@code benchmark:counter} rows. + * + * <p>Throughout this class, an <em>operation</em> is JMH's unit of work: a single invocation of a + * {@code @Benchmark}-annotated method. (One operation equals one invocation here because we don't use + * {@code @OperationsPerInvocation}.) JMH reports the timing score in {@code ns/op}, and these work + * counters are reported {@code PerOp} to match. + * + * <p>Each benchmark calls {@link #drainFrom} every invocation to accumulate the work deltas drained + * from {@link RaftClientBenchmarkContext}. The raw totals are private accumulators; what we report + * are the per-operation values from the {@code *PerOp()} methods (the quantity of interest), plus + * {@link #operations}. + * + * <p>JMH aggregates {@code Type.EVENTS} secondary results with {@code SUM} across all measurement + * data points i.e {@code forks x measurement iterations}. To make the <em>summary</em> row + * report the true per-operation value rather than that value multiplied by the data-point count, each + * method pre-divides by the data-point count obtained from {@link BenchmarkParams} in + * {@link #captureRunShape}. The SUM then reconstitutes the exact per-operation value (e.g. + * {@code logReadsPerOp = 1.0}) in the summary, for any {@code -f}/{@code -i} configuration. (The + * per-iteration console values are correspondingly a small fraction of the per-op value; read the + * summary row.) + * + * <p>The per-operation values are integer-exact and should be stable across a correct refactor of + * {@code KafkaRaftClient}: a flush count moving from 1 to 2 per operation is a behavioral diff, not + * measurement noise. The counters that are zero on a path (e.g. log flushes on a caught-up fetch) + * are the most useful tripwires, since zero is speed-independent. + */ +@State(Scope.Thread) +@AuxCounters(AuxCounters.Type.EVENTS) +public class KRaftBenchmarkingCounters { + private long logFlushesTotal; + private long logReadsTotal; + private long logTruncationsTotal; + private long rpcRequestsSentTotal; + private long rpcResponsesSentTotal; + private long quorumStateWritesTotal; + private long quorumStateReadsTotal; + + // Reported: the number of operations (i.e. @Benchmark method invocations) measured in the + // iteration, and the divisor for the per-operation values below. + public long operations; + + // The number of measurement data points JMH will SUM the per-op methods over, i.e. + // (forks x measurement iterations) for this run. Captured from BenchmarkParams so it tracks the + // actual run shape (including -f/-i overrides) rather than being hardcoded. + private double measurementDataPoints = 1.0; + + @Setup(Level.Trial) + public void captureRunShape(BenchmarkParams params) { + // forks() is 0 when forking is disabled (in-process), which is still one set of iterations. + int forks = Math.max(1, params.getForks()); + measurementDataPoints = (double) forks * params.getMeasurement().getCount(); + } + + @Setup(Level.Iteration) + public void reset() { + logFlushesTotal = 0; + logReadsTotal = 0; + logTruncationsTotal = 0; + rpcRequestsSentTotal = 0; + rpcResponsesSentTotal = 0; + quorumStateWritesTotal = 0; + quorumStateReadsTotal = 0; + } + + /** + * Accumulates this invocation's work deltas drained from {@code context} into these counters. + * {@code expectedRequest}/{@code expectedResponse}, if present, restrict the RPC request/response + * counts to that API key (e.g. {@code FETCH}); empty counts all. Review Comment: We should document that when `expectedRequest/expectedResponse` is empty, it means no outstanding requests/responses are expected at the end of a benchmarking invocation. ########## raft/src/testFixtures/java/org/apache/kafka/raft/RaftClientBenchmarkContext.java: ########## @@ -176,7 +219,21 @@ public int drainQuorumStateWrites() { return delta; } - public int drainRpcResponsesSent() { - return context.drainAllSentResponses(); + public int drainQuorumStateReads() { + int current = context.quorumStateReadCount(); + int delta = current - lastQuorumReads; + lastQuorumReads = current; + return delta; + } + + /** + * Number of responses sent since the last drain. If {@code apiKey} is present, only responses of + * that API key are counted; otherwise all responses are counted. + */ + public int drainRpcResponsesSent(Optional<ApiKeys> apiKey) { + if (apiKey.isEmpty()) { + return context.drainAllSentResponses(); + } Review Comment: If a benchmarked code has outstanding sent responses before calling this method, the benchmark writer should know what those are. What do you think about renaming this method to `maybeDrainSentRpcResponses(Optional<ApiKeys> apiKey)` and the below implementation: ``` if apiKey is present, drain for the specific api key supplied and store that number assert the sentResponses queue is empty return the stored number or 0 ``` The motivation for this implementation is similar to what is proposed for properly draining + collecting deltas on the send queue here: https://github.com/apache/kafka/pull/22669/changes/f9b46f94589e96a65e076c56979bdc8788982ec7..e34b656497d2b9c17a027272ab6b028fc436ca70#r3484997654 ########## jmh-benchmarks/src/main/java/org/apache/kafka/jmh/raft/KRaftBenchmarkingCounters.java: ########## @@ -0,0 +1,151 @@ +/* + * 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.kafka.jmh.raft; + +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.raft.RaftClientBenchmarkContext; + +import org.openjdk.jmh.annotations.AuxCounters; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.infra.BenchmarkParams; + +import java.util.Optional; + +/** + * Secondary, machine-independent work counters reported by the raft benchmarks alongside the timing + * score, as {@code benchmark:counter} rows. + * + * <p>Throughout this class, an <em>operation</em> is JMH's unit of work: a single invocation of a + * {@code @Benchmark}-annotated method. (One operation equals one invocation here because we don't use + * {@code @OperationsPerInvocation}.) JMH reports the timing score in {@code ns/op}, and these work + * counters are reported {@code PerOp} to match. + * + * <p>Each benchmark calls {@link #drainFrom} every invocation to accumulate the work deltas drained + * from {@link RaftClientBenchmarkContext}. The raw totals are private accumulators; what we report + * are the per-operation values from the {@code *PerOp()} methods (the quantity of interest), plus + * {@link #operations}. + * + * <p>JMH aggregates {@code Type.EVENTS} secondary results with {@code SUM} across all measurement + * data points i.e {@code forks x measurement iterations}. To make the <em>summary</em> row + * report the true per-operation value rather than that value multiplied by the data-point count, each + * method pre-divides by the data-point count obtained from {@link BenchmarkParams} in + * {@link #captureRunShape}. The SUM then reconstitutes the exact per-operation value (e.g. + * {@code logReadsPerOp = 1.0}) in the summary, for any {@code -f}/{@code -i} configuration. (The + * per-iteration console values are correspondingly a small fraction of the per-op value; read the + * summary row.) + * + * <p>The per-operation values are integer-exact and should be stable across a correct refactor of + * {@code KafkaRaftClient}: a flush count moving from 1 to 2 per operation is a behavioral diff, not + * measurement noise. The counters that are zero on a path (e.g. log flushes on a caught-up fetch) + * are the most useful tripwires, since zero is speed-independent. + */ +@State(Scope.Thread) +@AuxCounters(AuxCounters.Type.EVENTS) +public class KRaftBenchmarkingCounters { + private long logFlushesTotal; + private long logReadsTotal; + private long logTruncationsTotal; + private long rpcRequestsSentTotal; + private long rpcResponsesSentTotal; + private long quorumStateWritesTotal; + private long quorumStateReadsTotal; + + // Reported: the number of operations (i.e. @Benchmark method invocations) measured in the + // iteration, and the divisor for the per-operation values below. + public long operations; + + // The number of measurement data points JMH will SUM the per-op methods over, i.e. + // (forks x measurement iterations) for this run. Captured from BenchmarkParams so it tracks the + // actual run shape (including -f/-i overrides) rather than being hardcoded. + private double measurementDataPoints = 1.0; + + @Setup(Level.Trial) + public void captureRunShape(BenchmarkParams params) { + // forks() is 0 when forking is disabled (in-process), which is still one set of iterations. + int forks = Math.max(1, params.getForks()); + measurementDataPoints = (double) forks * params.getMeasurement().getCount(); + } Review Comment: Can you explain what this code is doing for my understanding? I thought this state is per thread. How does jmh reconcile multiple of these when forks > 1? What does `params.getMeasurement().getCount()` mean? ########## jmh-benchmarks/src/main/java/org/apache/kafka/jmh/raft/KRaftBenchmarkingCounters.java: ########## @@ -0,0 +1,151 @@ +/* + * 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.kafka.jmh.raft; + +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.raft.RaftClientBenchmarkContext; + +import org.openjdk.jmh.annotations.AuxCounters; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.infra.BenchmarkParams; + +import java.util.Optional; + +/** + * Secondary, machine-independent work counters reported by the raft benchmarks alongside the timing + * score, as {@code benchmark:counter} rows. + * + * <p>Throughout this class, an <em>operation</em> is JMH's unit of work: a single invocation of a + * {@code @Benchmark}-annotated method. (One operation equals one invocation here because we don't use + * {@code @OperationsPerInvocation}.) JMH reports the timing score in {@code ns/op}, and these work + * counters are reported {@code PerOp} to match. + * + * <p>Each benchmark calls {@link #drainFrom} every invocation to accumulate the work deltas drained + * from {@link RaftClientBenchmarkContext}. The raw totals are private accumulators; what we report + * are the per-operation values from the {@code *PerOp()} methods (the quantity of interest), plus + * {@link #operations}. + * + * <p>JMH aggregates {@code Type.EVENTS} secondary results with {@code SUM} across all measurement + * data points i.e {@code forks x measurement iterations}. To make the <em>summary</em> row + * report the true per-operation value rather than that value multiplied by the data-point count, each + * method pre-divides by the data-point count obtained from {@link BenchmarkParams} in + * {@link #captureRunShape}. The SUM then reconstitutes the exact per-operation value (e.g. + * {@code logReadsPerOp = 1.0}) in the summary, for any {@code -f}/{@code -i} configuration. (The + * per-iteration console values are correspondingly a small fraction of the per-op value; read the + * summary row.) + * + * <p>The per-operation values are integer-exact and should be stable across a correct refactor of + * {@code KafkaRaftClient}: a flush count moving from 1 to 2 per operation is a behavioral diff, not + * measurement noise. The counters that are zero on a path (e.g. log flushes on a caught-up fetch) + * are the most useful tripwires, since zero is speed-independent. + */ +@State(Scope.Thread) +@AuxCounters(AuxCounters.Type.EVENTS) +public class KRaftBenchmarkingCounters { + private long logFlushesTotal; + private long logReadsTotal; + private long logTruncationsTotal; + private long rpcRequestsSentTotal; + private long rpcResponsesSentTotal; + private long quorumStateWritesTotal; + private long quorumStateReadsTotal; + + // Reported: the number of operations (i.e. @Benchmark method invocations) measured in the + // iteration, and the divisor for the per-operation values below. + public long operations; + + // The number of measurement data points JMH will SUM the per-op methods over, i.e. + // (forks x measurement iterations) for this run. Captured from BenchmarkParams so it tracks the + // actual run shape (including -f/-i overrides) rather than being hardcoded. + private double measurementDataPoints = 1.0; + + @Setup(Level.Trial) + public void captureRunShape(BenchmarkParams params) { + // forks() is 0 when forking is disabled (in-process), which is still one set of iterations. + int forks = Math.max(1, params.getForks()); + measurementDataPoints = (double) forks * params.getMeasurement().getCount(); + } + + @Setup(Level.Iteration) + public void reset() { + logFlushesTotal = 0; + logReadsTotal = 0; + logTruncationsTotal = 0; + rpcRequestsSentTotal = 0; + rpcResponsesSentTotal = 0; + quorumStateWritesTotal = 0; + quorumStateReadsTotal = 0; + } + + /** + * Accumulates this invocation's work deltas drained from {@code context} into these counters. + * {@code expectedRequest}/{@code expectedResponse}, if present, restrict the RPC request/response + * counts to that API key (e.g. {@code FETCH}); empty counts all. + */ + public void drainFrom( + RaftClientBenchmarkContext context, + Optional<ApiKeys> expectedRequest, + Optional<ApiKeys> expectedResponse + ) { + logFlushesTotal += context.drainLogFlushes(); + logReadsTotal += context.drainLogReads(); + logTruncationsTotal += context.drainLogTruncations(); + rpcRequestsSentTotal += context.drainRpcRequestsSent(expectedRequest); Review Comment: From my previous comments, before this line, we should actually drain the `MockNetworkChannel#sendQueue` of outstanding requests, and then assert the queue is empty. ########## raft/src/testFixtures/java/org/apache/kafka/raft/RaftClientBenchmarkContext.java: ########## @@ -51,93 +70,106 @@ private RaftClientBenchmarkContext(RaftClientTestContext context, ReplicaKey oth * Unattached → Leader election. A single-voter quorum is rejected because such a node elects * itself at initialization, before any measured poll. */ - public static RaftClientBenchmarkContext unattached(int localId, int voterCount) throws Exception { + public static RaftClientBenchmarkContext unattached(int voterCount) throws Exception { + return unattached(voterCount, DEFAULT_KRAFT_VERSION, DEFAULT_RAFT_PROTOCOL); + } + + public static RaftClientBenchmarkContext unattached( + int voterCount, + KRaftVersion kraftVersion, + RaftProtocol raftProtocol + ) throws Exception { if (voterCount < 2) { throw new IllegalArgumentException("voterCount must be at least 2; a single voter self-elects at init"); } - return new RaftClientBenchmarkContext(buildContext(voterKeys(localId, voterCount)), null); + List<ReplicaKey> voterKeys = voterKeys(voterCount); + return new RaftClientBenchmarkContext(buildContext(voterKeys, kraftVersion, raftProtocol), voterKeys); + } + + public static RaftClientBenchmarkContext leader(int voterCount) throws Exception { + return leader(voterCount, DEFAULT_KRAFT_VERSION, DEFAULT_RAFT_PROTOCOL); } - public static RaftClientBenchmarkContext leader(int localId, int voterCount) throws Exception { - List<ReplicaKey> voterKeys = voterKeys(localId, voterCount); - RaftClientTestContext context = buildContext(voterKeys); + public static RaftClientBenchmarkContext leader( + int voterCount, + KRaftVersion kraftVersion, + RaftProtocol raftProtocol + ) throws Exception { + List<ReplicaKey> voterKeys = voterKeys(voterCount); + RaftClientTestContext context = buildContext(voterKeys, kraftVersion, raftProtocol); context.unattachedToLeader(); - return new RaftClientBenchmarkContext(context, voterKeys.get(1)); + return new RaftClientBenchmarkContext(context, voterKeys); } - /** {@code voterCount} voter keys, each with a random directory id, with {@code localId} first. */ - private static List<ReplicaKey> voterKeys(int localId, int voterCount) { + /** + * {@code voterCount} voter keys, each with a random directory id, with the local node first. + */ + private static List<ReplicaKey> voterKeys(int voterCount) { + int localId = randomReplicaId(); return IntStream.range(0, voterCount) .mapToObj(i -> ReplicaKey.of(localId + i, Uuid.randomUuid())) .collect(Collectors.toList()); } + private static int randomReplicaId() { + return ThreadLocalRandom.current().nextInt(1025); + } + /** - * Builds an unattached node in a KIP-1186 quorum of {@code voterKeys}, whose first entry is the - * local node. + * Builds an unattached node in a quorum of {@code voterKeys} (first entry is the local node) Review Comment: Technically there is no quorum at this point because there is no leader. It is more accurate to say: ``` Initializes a local, unattached node in a cluster of ... ``` ########## raft/src/testFixtures/java/org/apache/kafka/raft/RaftClientBenchmarkContext.java: ########## @@ -162,10 +194,21 @@ public int drainLogTruncations() { return delta; } - public int drainRpcRequestsSent() { - int current = channel.requestsSent(); - int delta = current - lastRequestsSent; - lastRequestsSent = current; + /** + * Number of requests sent since the last drain. If {@code apiKey} is present, only requests of + * that API key are counted; otherwise all requests are counted. + */ + public int drainRpcRequestsSent(Optional<ApiKeys> apiKey) { Review Comment: Let's rename methods to `get...Deltas`. Same with the methods in the `MockLog`/`MockQuorumStateStore`. This is not actually draining the request queue, so that is confusing. We can remove the `apiKey` as a parameter after addressing https://github.com/apache/kafka/pull/22669/changes/f9b46f94589e96a65e076c56979bdc8788982ec7..e34b656497d2b9c17a027272ab6b028fc436ca70#r3484997654. ########## raft/src/testFixtures/java/org/apache/kafka/raft/MockNetworkChannel.java: ########## @@ -35,8 +35,8 @@ public class MockNetworkChannel implements NetworkChannel { private final List<RequestEntry> sendQueue = new ArrayList<>(); private final Map<Integer, CompletableFuture<RaftResponse.Inbound>> awaitingResponse = new HashMap<>(); - // Cumulative count of outbound requests sent, used by the JMH raft benchmarks. private int requestsSent = 0; + private final Map<Short, Integer> requestsSentByApiKey = new HashMap<>(); Review Comment: Take a look at `MockNetworkChannel#drainSentRequests(apiKeyFilter)`. We should not need to store another map in this class to drain requests at the end of a benchmark invocation. We should call the method above in `drainFrom`. It is the benchmark writer's responsibility to know, at the end of a benchmark method invocation, what requests/responses are in-flight, and to drain them. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
