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     protected int dataSize = 0;
  30
  31     public AbstractReplicatedLogImpl(long snapshotIndex,
  32         long snapshotTerm, List<ReplicatedLogEntry> unAppliedEntries) {
  33         this.snapshotIndex = snapshotIndex;
  34         this.snapshotTerm = snapshotTerm;
  35         this.journal = new ArrayList<>(unAppliedEntries);
  36     }
  37
  38     public AbstractReplicatedLogImpl() {
  39         this.journal = new ArrayList<>();
  40     }
  41
  42     protected int adjustedIndex(long logEntryIndex) {
  43         if(snapshotIndex < 0){
  44             return (int) logEntryIndex;
  45         }
  46         return (int) (logEntryIndex - (snapshotIndex + 1));
  47     }
  48
  49     @Override
  50     public ReplicatedLogEntry get(long logEntryIndex) {
  51         int adjustedIndex = adjustedIndex(logEntryIndex);
  52
  53         if (adjustedIndex < 0 || adjustedIndex >= journal.size()) {
  54             // physical index should be less than list size and >= 0
  55             return null;
  56         }
  57
  58         return journal.get(adjustedIndex);
  59     }
  60
  61     @Override
  62     public ReplicatedLogEntry last() {
  63         if (journal.isEmpty()) {
  64             return null;
  65         }
  66         // get the last entry directly from the physical index
  67         return journal.get(journal.size() - 1);
  68     }
  69
  70     @Override
  71     public long lastIndex() {
  72         if (journal.isEmpty()) {
  73             // it can happen that after snapshot, all the entries of the
  74             // journal are trimmed till lastApplied, so lastIndex = snapshotIndex
  75             return snapshotIndex;
  76         }
  77         return last().getIndex();
  78     }
  79
  80     @Override
  81     public long lastTerm() {
  82         if (journal.isEmpty()) {
  83             // it can happen that after snapshot, all the entries of the
  84             // journal are trimmed till lastApplied, so lastTerm = snapshotTerm
  85             return snapshotTerm;
  86         }
  87         return last().getTerm();
  88     }
  89
  90     @Override
  91     public void removeFrom(long logEntryIndex) {
  92         int adjustedIndex = adjustedIndex(logEntryIndex);
  93         if (adjustedIndex < 0 || adjustedIndex >= journal.size()) {
  94             // physical index should be less than list size and >= 0
  95             return;
  96         }
  97         journal.subList(adjustedIndex , journal.size()).clear();
  98     }
  99
 100     @Override
 101     public void append(ReplicatedLogEntry replicatedLogEntry) {
 102         journal.add(replicatedLogEntry);
 103     }
 104
 105     @Override
 106     public void increaseJournalLogCapacity(int amount) {
 107         journal.ensureCapacity(journal.size() + amount);
 108     }
 109
 110     @Override
 111     public List<ReplicatedLogEntry> getFrom(long logEntryIndex) {
 112         return getFrom(logEntryIndex, journal.size());
 113     }
 114
 115     @Override
 116     public List<ReplicatedLogEntry> getFrom(long logEntryIndex, int max) {
 117         int adjustedIndex = adjustedIndex(logEntryIndex);
 118         int size = journal.size();
 119         List<ReplicatedLogEntry> entries = new ArrayList<>(100);
 120         if (adjustedIndex >= 0 && adjustedIndex < size) {
 121             // physical index should be less than list size and >= 0
 122             int maxIndex = adjustedIndex + max;
 123             if(maxIndex > size){
 124                 maxIndex = size;
 125             }
 126             entries.addAll(journal.subList(adjustedIndex, maxIndex));
 127         }
 128         return entries;
 129     }
 130
 131
 132     @Override
 133     public long size() {
 134        return journal.size();
 135     }
 136
 137     @Override
 138     public boolean isPresent(long logEntryIndex) {
 139         if (logEntryIndex > lastIndex()) {
 140             // if the request logical index is less than the last present in the list
 141             return false;
 142         }
 143         int adjustedIndex = adjustedIndex(logEntryIndex);
 144         return (adjustedIndex >= 0);
 145     }
 146
 147     @Override
 148     public boolean isInSnapshot(long logEntryIndex) {
 149         return logEntryIndex <= snapshotIndex && snapshotIndex != -1;
 150     }
 151
 152     @Override
 153     public long getSnapshotIndex() {
 154         return snapshotIndex;
 155     }
 156
 157     @Override
 158     public long getSnapshotTerm() {
 159         return snapshotTerm;
 160     }
 161
 162     @Override
 163     public abstract void appendAndPersist(ReplicatedLogEntry replicatedLogEntry);
 164
 165     @Override
 166     public abstract void removeFromAndPersist(long index);
 167
 168     @Override
 169     public void setSnapshotIndex(long snapshotIndex) {
 170         this.snapshotIndex = snapshotIndex;
 171     }
 172
 173     @Override
 174     public void setSnapshotTerm(long snapshotTerm) {
 175         this.snapshotTerm = snapshotTerm;
 176     }
 177
 178     @Override
 179     public void clear(int startIndex, int endIndex) {
 180         journal.subList(startIndex, endIndex).clear();
 181     }
 182
 183     @Override
 184     public void snapshotPreCommit(long snapshotCapturedIndex, long snapshotCapturedTerm) {
 185         snapshottedJournal = new ArrayList<>(journal.size());
 186
 187         snapshottedJournal.addAll(journal.subList(0, (int)(snapshotCapturedIndex - snapshotIndex)));
 188         clear(0, (int) (snapshotCapturedIndex - snapshotIndex));
 189
 190         previousSnapshotIndex = snapshotIndex;
 191         setSnapshotIndex(snapshotCapturedIndex);
 192
 193         previousSnapshotTerm = snapshotTerm;
 194         setSnapshotTerm(snapshotCapturedTerm);
 195     }
 196
 197     @Override
 198     public void snapshotCommit() {
 199         snapshottedJournal = null;
 200         previousSnapshotIndex = -1;
 201         previousSnapshotTerm = -1;
 202         dataSize = 0;
 203     }
 204
 205     @Override
 206     public void snapshotRollback() {
 207         snapshottedJournal.addAll(journal);
 208         journal = snapshottedJournal;
 209         snapshottedJournal = null;
 210
 211         snapshotIndex = previousSnapshotIndex;
 212         previousSnapshotIndex = -1;
 213
 214         snapshotTerm = previousSnapshotTerm;
 215         previousSnapshotTerm = -1;
 216     }
 217 }