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