Adaptive batch size

resetius · Dec 1, 2023 · c165f2e · c165f2e
1 parent 9245b3e
commit c165f2e
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 1 deletion.
diff --git a/src/raft.cpp b/src/raft.cpp
@@ -211,11 +211,13 @@ void TRaft::OnAppendEntries(TMessageHolder<TAppendEntriesResponse> message) {
             .MergeNextIndex({{nodeId, message->MatchIndex+1}})
             .CommitAdvance(Nservers, State->Log.size(), *State)
             .MergeRpcDue({{nodeId, ITimeSource::Time{}}});
+        VolatileState->BatchSize[nodeId] = 1024;
     } else {
         (*VolatileState)
             .MergeNextIndex({{nodeId, std::max((uint64_t)1, VolatileState->NextIndex[nodeId]-1)}})
             //.MergeNextIndex({{nodeId, 1}})
             .MergeRpcDue({{nodeId, ITimeSource::Time{}}});
+        VolatileState->BatchSize[nodeId] = 1;
     }
 }
 
@@ -231,7 +233,7 @@ TMessageHolder<TRequestVoteRequest> TRaft::CreateVote(uint32_t nodeId) {
 }
 
 TMessageHolder<TAppendEntriesRequest> TRaft::CreateAppendEntries(uint32_t nodeId) {
-    static constexpr int batchSize = 128;
+    int batchSize = std::max(1, VolatileState->BatchSize[nodeId]);
     auto prevIndex = VolatileState->NextIndex[nodeId] - 1;
     auto lastIndex = std::min(prevIndex+batchSize, (uint64_t)State->Log.size());
     if (VolatileState->MatchIndex[nodeId]+1 < VolatileState->NextIndex[nodeId]) {

diff --git a/src/raft.h b/src/raft.h
@@ -43,6 +43,7 @@ struct TVolatileState {
     std::unordered_set<uint32_t> Votes;
     std::unordered_map<uint32_t, ITimeSource::Time> HeartbeatDue;
     std::unordered_map<uint32_t, ITimeSource::Time> RpcDue;
+    std::unordered_map<uint32_t, int> BatchSize;
     ITimeSource::Time ElectionDue;
 
     TVolatileState& SetVotes(std::unordered_set<uint32_t>& votes);