Bug 2787: Batch AppendEntries to speed up follower sync

[controller.git] / opendaylight / md-sal / sal-akka-raft / src / main / java / org / opendaylight / controller / cluster / raft / AbstractReplicatedLogImpl.java

diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/AbstractReplicatedLogImpl.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/AbstractReplicatedLogImpl.java
index b4b2afbc4ad602ccd1bc9f50da8641cd0b05f605..c245206f641f3a4ff31da8608076f1c3d68cb4f6 100644 (file)
--- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/AbstractReplicatedLogImpl.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/AbstractReplicatedLogImpl.java
@@ -124,20 +124,43 @@ public abstract class AbstractReplicatedLogImpl implements ReplicatedLog {
 
     @Override
     public List<ReplicatedLogEntry> getFrom(long logEntryIndex) {
-        return getFrom(logEntryIndex, journal.size());
+        return getFrom(logEntryIndex, journal.size(), NO_MAX_SIZE);
     }
 
     @Override
-    public List<ReplicatedLogEntry> getFrom(long logEntryIndex, int max) {
+    public List<ReplicatedLogEntry> getFrom(long logEntryIndex, int maxEntries, long maxDataSize) {
         int adjustedIndex = adjustedIndex(logEntryIndex);
         int size = journal.size();
         if (adjustedIndex >= 0 && adjustedIndex < size) {
             // physical index should be less than list size and >= 0
-            int maxIndex = adjustedIndex + max;
+            int maxIndex = adjustedIndex + maxEntries;
             if(maxIndex > size){
                 maxIndex = size;
             }
-            return new ArrayList<>(journal.subList(adjustedIndex, maxIndex));
+
+            if(maxDataSize == NO_MAX_SIZE) {
+                return new ArrayList<>(journal.subList(adjustedIndex, maxIndex));
+            } else {
+                List<ReplicatedLogEntry> retList = new ArrayList<>(maxIndex - adjustedIndex);
+                long totalSize = 0;
+                for(int i = adjustedIndex; i < maxIndex; i++) {
+                    ReplicatedLogEntry entry = journal.get(i);
+                    totalSize += entry.size();
+                    if(totalSize <= maxDataSize) {
+                        retList.add(entry);
+                    } else {
+                        if(retList.isEmpty()) {
+                            // Edge case - the first entry's size exceeds the threshold. We need to return
+                            // at least the first entry so add it here.
+                            retList.add(entry);
+                        }
+
+                        break;
+                    }
+                }
+
+                return retList;
+            }
         } else {
             return Collections.emptyList();
         }