X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-akka-raft%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fraft%2FAbstractReplicatedLogImpl.java;h=c27ff373865069df16b08eae829e1722ff74a5f8;hp=b436bce50061f98d0362ce3df4336c4279977d63;hb=68fb550b416dddd0a50e0110add0a4ae9b706758;hpb=b10d77375b5a290143106180f1583ea4e18f8478 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 b436bce500..c27ff37386 100644 --- 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 @@ -7,9 +7,10 @@ */ package org.opendaylight.controller.cluster.raft; -import com.google.protobuf.ByteString; - +import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; import java.util.ArrayList; +import java.util.Collections; import java.util.List; /** @@ -18,33 +19,35 @@ import java.util.List; */ public abstract class AbstractReplicatedLogImpl implements ReplicatedLog { - protected List journal; - protected ByteString snapshot; - protected long snapshotIndex = -1; - protected long snapshotTerm = -1; + // We define this as ArrayList so we can use ensureCapacity. + private ArrayList journal; + + private long snapshotIndex = -1; + private long snapshotTerm = -1; // to be used for rollback during save snapshot failure - protected List snapshottedJournal; - protected ByteString previousSnapshot; - protected long previousSnapshotIndex = -1; - protected long previousSnapshotTerm = -1; + private ArrayList snapshottedJournal; + private long previousSnapshotIndex = -1; + private long previousSnapshotTerm = -1; + private int dataSize = 0; - public AbstractReplicatedLogImpl(ByteString state, long snapshotIndex, + public AbstractReplicatedLogImpl(long snapshotIndex, long snapshotTerm, List unAppliedEntries) { - this.snapshot = state; this.snapshotIndex = snapshotIndex; this.snapshotTerm = snapshotTerm; this.journal = new ArrayList<>(unAppliedEntries); - } + for(ReplicatedLogEntry entry: journal) { + dataSize += entry.size(); + } + } public AbstractReplicatedLogImpl() { - this.snapshot = null; - this.journal = new ArrayList<>(); + this(-1L, -1L, Collections.emptyList()); } protected int adjustedIndex(long logEntryIndex) { - if(snapshotIndex < 0){ + if (snapshotIndex < 0) { return (int) logEntryIndex; } return (int) (logEntryIndex - (snapshotIndex + 1)); @@ -92,18 +95,31 @@ public abstract class AbstractReplicatedLogImpl implements ReplicatedLog { } @Override - public void removeFrom(long logEntryIndex) { + public long removeFrom(long logEntryIndex) { int adjustedIndex = adjustedIndex(logEntryIndex); if (adjustedIndex < 0 || adjustedIndex >= journal.size()) { // physical index should be less than list size and >= 0 - return; + return -1; + } + + for(int i = adjustedIndex; i < journal.size(); i++) { + dataSize -= journal.get(i).size(); } + journal.subList(adjustedIndex , journal.size()).clear(); + + return adjustedIndex; } @Override public void append(ReplicatedLogEntry replicatedLogEntry) { journal.add(replicatedLogEntry); + dataSize += replicatedLogEntry.size(); + } + + @Override + public void increaseJournalLogCapacity(int amount) { + journal.ensureCapacity(journal.size() + amount); } @Override @@ -115,24 +131,28 @@ public abstract class AbstractReplicatedLogImpl implements ReplicatedLog { public List getFrom(long logEntryIndex, int max) { int adjustedIndex = adjustedIndex(logEntryIndex); int size = journal.size(); - List entries = new ArrayList<>(100); if (adjustedIndex >= 0 && adjustedIndex < size) { // physical index should be less than list size and >= 0 int maxIndex = adjustedIndex + max; if(maxIndex > size){ maxIndex = size; } - entries.addAll(journal.subList(adjustedIndex, maxIndex)); + return new ArrayList<>(journal.subList(adjustedIndex, maxIndex)); + } else { + return Collections.emptyList(); } - return entries; } - @Override public long size() { return journal.size(); } + @Override + public int dataSize() { + return dataSize; + } + @Override public boolean isPresent(long logEntryIndex) { if (logEntryIndex > lastIndex()) { @@ -148,11 +168,6 @@ public abstract class AbstractReplicatedLogImpl implements ReplicatedLog { return logEntryIndex <= snapshotIndex && snapshotIndex != -1; } - @Override - public ByteString getSnapshot() { - return snapshot; - } - @Override public long getSnapshotIndex() { return snapshotIndex; @@ -179,21 +194,21 @@ public abstract class AbstractReplicatedLogImpl implements ReplicatedLog { this.snapshotTerm = snapshotTerm; } - @Override - public void setSnapshot(ByteString snapshot) { - this.snapshot = snapshot; - } - @Override public void clear(int startIndex, int endIndex) { journal.subList(startIndex, endIndex).clear(); } @Override - public void snapshotPreCommit(ByteString snapshot, long snapshotCapturedIndex, long snapshotCapturedTerm) { + public void snapshotPreCommit(long snapshotCapturedIndex, long snapshotCapturedTerm) { + Preconditions.checkArgument(snapshotCapturedIndex >= snapshotIndex, + "snapshotCapturedIndex must be greater than or equal to snapshotIndex"); + snapshottedJournal = new ArrayList<>(journal.size()); - snapshottedJournal.addAll(journal.subList(0, (int)(snapshotCapturedIndex - snapshotIndex))); + List snapshotJournalEntries = journal.subList(0, (int) (snapshotCapturedIndex - snapshotIndex)); + + snapshottedJournal.addAll(snapshotJournalEntries); clear(0, (int) (snapshotCapturedIndex - snapshotIndex)); previousSnapshotIndex = snapshotIndex; @@ -201,24 +216,24 @@ public abstract class AbstractReplicatedLogImpl implements ReplicatedLog { previousSnapshotTerm = snapshotTerm; setSnapshotTerm(snapshotCapturedTerm); - - previousSnapshot = getSnapshot(); - setSnapshot(snapshot); } @Override public void snapshotCommit() { - snapshottedJournal.clear(); snapshottedJournal = null; previousSnapshotIndex = -1; previousSnapshotTerm = -1; - previousSnapshot = null; + dataSize = 0; + // need to recalc the datasize based on the entries left after precommit. + for(ReplicatedLogEntry logEntry : journal) { + dataSize += logEntry.size(); + } + } @Override public void snapshotRollback() { snapshottedJournal.addAll(journal); - journal.clear(); journal = snapshottedJournal; snapshottedJournal = null; @@ -227,9 +242,10 @@ public abstract class AbstractReplicatedLogImpl implements ReplicatedLog { snapshotTerm = previousSnapshotTerm; previousSnapshotTerm = -1; + } - snapshot = previousSnapshot; - previousSnapshot = null; - + @VisibleForTesting + ReplicatedLogEntry getAtPhysicalIndex(int index) { + return journal.get(index); } }