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

   1 /*
   2  * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.cluster.raft;
   9
  10 import java.util.ArrayList;
  11 import java.util.List;
  12
  13 /**
  14  * Abstract class handling the mapping of
  15  * logical LogEntry Index and the physical list index.
  16  */
  17 public abstract class AbstractReplicatedLogImpl implements ReplicatedLog {
  18
  19     // We define this as ArrayList so we can use ensureCapacity.
  20     protected ArrayList<ReplicatedLogEntry> journal;
  21
  22     protected long snapshotIndex = -1;
  23     protected long snapshotTerm = -1;
  24
  25     // to be used for rollback during save snapshot failure
  26     protected ArrayList<ReplicatedLogEntry> snapshottedJournal;
  27     protected long previousSnapshotIndex = -1;
  28     protected long previousSnapshotTerm = -1;
  29
  30     public AbstractReplicatedLogImpl(long snapshotIndex,
  31         long snapshotTerm, List<ReplicatedLogEntry> unAppliedEntries) {
  32         this.snapshotIndex = snapshotIndex;
  33         this.snapshotTerm = snapshotTerm;
  34         this.journal = new ArrayList<>(unAppliedEntries);
  35     }
  36
  37     public AbstractReplicatedLogImpl() {
  38         this.journal = new ArrayList<>();
  39     }
  40
  41     protected int adjustedIndex(long logEntryIndex) {
  42         if(snapshotIndex < 0){
  43             return (int) logEntryIndex;
  44         }
  45         return (int) (logEntryIndex - (snapshotIndex + 1));
  46     }
  47
  48     @Override
  49     public ReplicatedLogEntry get(long logEntryIndex) {
  50         int adjustedIndex = adjustedIndex(logEntryIndex);
  51
  52         if (adjustedIndex < 0 || adjustedIndex >= journal.size()) {
  53             // physical index should be less than list size and >= 0
  54             return null;
  55         }
  56
  57         return journal.get(adjustedIndex);
  58     }
  59
  60     @Override
  61     public ReplicatedLogEntry last() {
  62         if (journal.isEmpty()) {
  63             return null;
  64         }
  65         // get the last entry directly from the physical index
  66         return journal.get(journal.size() - 1);
  67     }
  68
  69     @Override
  70     public long lastIndex() {
  71         if (journal.isEmpty()) {
  72             // it can happen that after snapshot, all the entries of the
  73             // journal are trimmed till lastApplied, so lastIndex = snapshotIndex
  74             return snapshotIndex;
  75         }
  76         return last().getIndex();
  77     }
  78
  79     @Override
  80     public long lastTerm() {
  81         if (journal.isEmpty()) {
  82             // it can happen that after snapshot, all the entries of the
  83             // journal are trimmed till lastApplied, so lastTerm = snapshotTerm
  84             return snapshotTerm;
  85         }
  86         return last().getTerm();
  87     }
  88
  89     @Override
  90     public void removeFrom(long logEntryIndex) {
  91         int adjustedIndex = adjustedIndex(logEntryIndex);
  92         if (adjustedIndex < 0 || adjustedIndex >= journal.size()) {
  93             // physical index should be less than list size and >= 0
  94             return;
  95         }
  96         journal.subList(adjustedIndex , journal.size()).clear();
  97     }
  98
  99     @Override
 100     public void append(ReplicatedLogEntry replicatedLogEntry) {
 101         journal.add(replicatedLogEntry);
 102     }
 103
 104     @Override
 105     public void increaseJournalLogCapacity(int amount) {
 106         journal.ensureCapacity(journal.size() + amount);
 107     }
 108
 109     @Override
 110     public List<ReplicatedLogEntry> getFrom(long logEntryIndex) {
 111         return getFrom(logEntryIndex, journal.size());
 112     }
 113
 114     @Override
 115     public List<ReplicatedLogEntry> getFrom(long logEntryIndex, int max) {
 116         int adjustedIndex = adjustedIndex(logEntryIndex);
 117         int size = journal.size();
 118         List<ReplicatedLogEntry> entries = new ArrayList<>(100);
 119         if (adjustedIndex >= 0 && adjustedIndex < size) {
 120             // physical index should be less than list size and >= 0
 121             int maxIndex = adjustedIndex + max;
 122             if(maxIndex > size){
 123                 maxIndex = size;
 124             }
 125             entries.addAll(journal.subList(adjustedIndex, maxIndex));
 126         }
 127         return entries;
 128     }
 129
 130
 131     @Override
 132     public long size() {
 133        return journal.size();
 134     }
 135
 136     @Override
 137     public boolean isPresent(long logEntryIndex) {
 138         if (logEntryIndex > lastIndex()) {
 139             // if the request logical index is less than the last present in the list
 140             return false;
 141         }
 142         int adjustedIndex = adjustedIndex(logEntryIndex);
 143         return (adjustedIndex >= 0);
 144     }
 145
 146     @Override
 147     public boolean isInSnapshot(long logEntryIndex) {
 148         return logEntryIndex <= snapshotIndex && snapshotIndex != -1;
 149     }
 150
 151     @Override
 152     public long getSnapshotIndex() {
 153         return snapshotIndex;
 154     }
 155
 156     @Override
 157     public long getSnapshotTerm() {
 158         return snapshotTerm;
 159     }
 160
 161     @Override
 162     public abstract void appendAndPersist(ReplicatedLogEntry replicatedLogEntry);
 163
 164     @Override
 165     public abstract void removeFromAndPersist(long index);
 166
 167     @Override
 168     public void setSnapshotIndex(long snapshotIndex) {
 169         this.snapshotIndex = snapshotIndex;
 170     }
 171
 172     @Override
 173     public void setSnapshotTerm(long snapshotTerm) {
 174         this.snapshotTerm = snapshotTerm;
 175     }
 176
 177     @Override
 178     public void clear(int startIndex, int endIndex) {
 179         journal.subList(startIndex, endIndex).clear();
 180     }
 181
 182     @Override
 183     public void snapshotPreCommit(long snapshotCapturedIndex, long snapshotCapturedTerm) {
 184         snapshottedJournal = new ArrayList<>(journal.size());
 185
 186         snapshottedJournal.addAll(journal.subList(0, (int)(snapshotCapturedIndex - snapshotIndex)));
 187         clear(0, (int) (snapshotCapturedIndex - snapshotIndex));
 188
 189         previousSnapshotIndex = snapshotIndex;
 190         setSnapshotIndex(snapshotCapturedIndex);
 191
 192         previousSnapshotTerm = snapshotTerm;
 193         setSnapshotTerm(snapshotCapturedTerm);
 194     }
 195
 196     @Override
 197     public void snapshotCommit() {
 198         snapshottedJournal = null;
 199         previousSnapshotIndex = -1;
 200         previousSnapshotTerm = -1;
 201     }
 202
 203     @Override
 204     public void snapshotRollback() {
 205         snapshottedJournal.addAll(journal);
 206         journal = snapshottedJournal;
 207         snapshottedJournal = null;
 208
 209         snapshotIndex = previousSnapshotIndex;
 210         previousSnapshotIndex = -1;
 211
 212         snapshotTerm = previousSnapshotTerm;
 213         previousSnapshotTerm = -1;
 214     }
 215 }