perf: add RAFT_D_STATE_MACHINE_EXEC

dtprj · Jun 10, 2024 · deb902d · deb902d
1 parent 69e9a84
commit deb902d
Show file tree

Hide file tree

Showing 3 changed files with 16 additions and 0 deletions.
diff --git a/benchmark/src/main/java/com/github/dtprj/dongting/bench/raft/RaftPerfCallback.java b/benchmark/src/main/java/com/github/dtprj/dongting/bench/raft/RaftPerfCallback.java
@@ -52,6 +52,7 @@ public class RaftPerfCallback extends SimplePerfCallback {
     private final Summary raftReplicateRpcTime;
     private final Summary raftReplicateRpcItems;
     private final Summary raftReplicateRpcBytes;
+    private final Summary raftStateMachineExec;
 
     public RaftPerfCallback(boolean useNanos, String prefix) {
         super(useNanos);
@@ -81,6 +82,7 @@ public RaftPerfCallback(boolean useNanos, String prefix) {
         this.raftReplicateRpcTime = createSummary(prefix + "raft_replicate_rpc_time");
         this.raftReplicateRpcItems = createSummary(prefix + "raft_replicate_rpc_items");
         this.raftReplicateRpcBytes = createSummary(prefix + "raft_replicate_rpc_bytes");
+        this.raftStateMachineExec = createSummary(prefix + "raft_state_machine_exec");
     }
 
     @Override
@@ -152,6 +154,9 @@ public void onEvent(int perfType, long costTime, int count, long sum) {
                 raftReplicateRpcItems.observe(count);
                 raftReplicateRpcBytes.observe(sum);
                 break;
+            case RAFT_D_STATE_MACHINE_EXEC:
+                raftStateMachineExec.observe(costTime);
+                break;
         }
     }
 
@@ -182,6 +187,7 @@ public void printStats() {
         printTime(raftReplicateRpcTime);
         printValue(raftReplicateRpcItems);
         printValue(raftReplicateRpcBytes);
+        printTime(raftStateMachineExec);
 
         if (accept(FIBER_D_POLL) && accept(FIBER_D_WORK)) {
             double total = fiberPoll.get().sum + fiberWork.get().sum;

diff --git a/client/src/main/java/com/github/dtprj/dongting/net/PerfConsts.java b/client/src/main/java/com/github/dtprj/dongting/net/PerfConsts.java
@@ -47,4 +47,5 @@ public interface PerfConsts {
     int RAFT_D_IDX_WRITE = 210;
     int RAFT_D_IDX_WRITE_AND_FORCE = 211;
     int RAFT_D_REPLICATE_RPC = 212;
+    int RAFT_D_STATE_MACHINE_EXEC = 213;
 }
diff --git a/server/src/main/java/com/github/dtprj/dongting/raft/impl/ApplyManager.java b/server/src/main/java/com/github/dtprj/dongting/raft/impl/ApplyManager.java
@@ -17,6 +17,7 @@
 
 import com.github.dtprj.dongting.codec.ByteArrayEncoder;
 import com.github.dtprj.dongting.common.DtUtil;
+import com.github.dtprj.dongting.common.PerfCallback;
 import com.github.dtprj.dongting.common.Timestamp;
 import com.github.dtprj.dongting.fiber.Fiber;
 import com.github.dtprj.dongting.fiber.FiberCondition;
@@ -27,6 +28,7 @@
 import com.github.dtprj.dongting.log.BugLog;
 import com.github.dtprj.dongting.log.DtLog;
 import com.github.dtprj.dongting.log.DtLogs;
+import com.github.dtprj.dongting.net.PerfConsts;
 import com.github.dtprj.dongting.raft.RaftException;
 import com.github.dtprj.dongting.raft.server.LogItem;
 import com.github.dtprj.dongting.raft.server.RaftExecTimeoutException;
@@ -62,10 +64,13 @@ public class ApplyManager {
     private long initCommitIndex;
     private boolean initFutureComplete = false;
 
+    private final PerfCallback perfCallback;
+
     public ApplyManager(GroupComponents gc) {
         this.ts = gc.getRaftStatus().getTs();
         this.raftStatus = gc.getRaftStatus();
         this.gc = gc;
+        this.perfCallback = gc.getGroupConfig().getPerfCallback();
     }
 
     public void postInit() {
@@ -101,7 +106,9 @@ public void execRead(long index, RaftTask rt) {
                 future.completeExceptionally(new RaftExecTimeoutException("timeout "
                         + input.getDeadline().getTimeout(TimeUnit.MILLISECONDS) + "ms"));
             }
+            long t = perfCallback.takeTime(PerfConsts.RAFT_D_STATE_MACHINE_EXEC);
             Object r = stateMachine.exec(index, input);
+            perfCallback.fireTime(PerfConsts.RAFT_D_STATE_MACHINE_EXEC, t);
             future.complete(new RaftOutput(index, r));
         } catch (Throwable e) {
             log.error("exec read failed. {}", e);
@@ -115,7 +122,9 @@ public void execRead(long index, RaftTask rt) {
     private void execNormalWrite(long index, RaftTask rt) {
         try {
             RaftInput input = rt.getInput();
+            long t = perfCallback.takeTime(PerfConsts.RAFT_D_STATE_MACHINE_EXEC);
             Object r = stateMachine.exec(index, input);
+            perfCallback.fireTime(PerfConsts.RAFT_D_STATE_MACHINE_EXEC, t);
             CompletableFuture<RaftOutput> future = rt.getFuture();
             if (future != null) {
                 future.complete(new RaftOutput(index, r));