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

   1 /*
   2  * Copyright (c) 2016 2015 Brocade Communications 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.behaviors;
   9
  10 import com.google.protobuf.ByteString;
  11 import java.util.Arrays;
  12 import org.slf4j.Logger;
  13 import org.slf4j.LoggerFactory;
  14
  15 /**
  16  * Encapsulates the leader state and logic for sending snapshot chunks to a follower.
  17  */
  18 public final class LeaderInstallSnapshotState {
  19     private static final Logger LOG = LoggerFactory.getLogger(LeaderInstallSnapshotState.class);
  20
  21     // The index of the first chunk that is sent when installing a snapshot
  22     static final int FIRST_CHUNK_INDEX = 1;
  23
  24     // The index that the follower should respond with if it needs the install snapshot to be reset
  25     static final int INVALID_CHUNK_INDEX = -1;
  26
  27     // This would be passed as the hash code of the last chunk when sending the first chunk
  28     static final int INITIAL_LAST_CHUNK_HASH_CODE = -1;
  29
  30     private int snapshotChunkSize;
  31     private final ByteString snapshotBytes;
  32     private final String logName;
  33     private int offset = 0;
  34     // the next snapshot chunk is sent only if the replyReceivedForOffset matches offset
  35     private int replyReceivedForOffset;
  36     // if replyStatus is false, the previous chunk is attempted
  37     private boolean replyStatus = false;
  38     private int chunkIndex;
  39     private final int totalChunks;
  40     private int lastChunkHashCode = INITIAL_LAST_CHUNK_HASH_CODE;
  41     private int nextChunkHashCode = INITIAL_LAST_CHUNK_HASH_CODE;
  42
  43     LeaderInstallSnapshotState(ByteString snapshotBytes, int snapshotChunkSize, String logName) {
  44         this.snapshotChunkSize = snapshotChunkSize;
  45         this.snapshotBytes = snapshotBytes;
  46         this.logName = logName;
  47         int size = snapshotBytes.size();
  48         totalChunks = size / snapshotChunkSize +
  49                 (size % snapshotChunkSize > 0 ? 1 : 0);
  50
  51         LOG.debug("{}: Snapshot {} bytes, total chunks to send: {}", logName, size, totalChunks);
  52
  53         replyReceivedForOffset = -1;
  54         chunkIndex = FIRST_CHUNK_INDEX;
  55     }
  56
  57     ByteString getSnapshotBytes() {
  58         return snapshotBytes;
  59     }
  60
  61     int incrementOffset() {
  62         if(replyStatus) {
  63             // if prev chunk failed, we would want to sent the same chunk again
  64             offset = offset + snapshotChunkSize;
  65         }
  66         return offset;
  67     }
  68
  69     int incrementChunkIndex() {
  70         if (replyStatus) {
  71             // if prev chunk failed, we would want to sent the same chunk again
  72             chunkIndex =  chunkIndex + 1;
  73         }
  74         return chunkIndex;
  75     }
  76
  77     int getChunkIndex() {
  78         return chunkIndex;
  79     }
  80
  81     int getTotalChunks() {
  82         return totalChunks;
  83     }
  84
  85     boolean canSendNextChunk() {
  86         // we only send a false if a chunk is sent but we have not received a reply yet
  87         return replyReceivedForOffset == offset;
  88     }
  89
  90     boolean isLastChunk(int index) {
  91         return totalChunks == index;
  92     }
  93
  94     void markSendStatus(boolean success) {
  95         if (success) {
  96             // if the chunk sent was successful
  97             replyReceivedForOffset = offset;
  98             replyStatus = true;
  99             lastChunkHashCode = nextChunkHashCode;
 100         } else {
 101             // if the chunk sent was failure
 102             replyReceivedForOffset = offset;
 103             replyStatus = false;
 104         }
 105     }
 106
 107     byte[] getNextChunk() {
 108         int snapshotLength = getSnapshotBytes().size();
 109         int start = incrementOffset();
 110         int size = snapshotChunkSize;
 111         if (snapshotChunkSize > snapshotLength) {
 112             size = snapshotLength;
 113         } else if (start + snapshotChunkSize > snapshotLength) {
 114             size = snapshotLength - start;
 115         }
 116
 117         byte[] nextChunk = new byte[size];
 118         getSnapshotBytes().copyTo(nextChunk, start, 0, size);
 119         nextChunkHashCode = Arrays.hashCode(nextChunk);
 120
 121         LOG.debug("{}: Next chunk: total length={}, offset={}, size={}, hashCode={}", logName,
 122                 snapshotLength, start, size, nextChunkHashCode);
 123         return nextChunk;
 124     }
 125
 126     /**
 127      * reset should be called when the Follower needs to be sent the snapshot from the beginning
 128      */
 129     void reset(){
 130         offset = 0;
 131         replyStatus = false;
 132         replyReceivedForOffset = offset;
 133         chunkIndex = FIRST_CHUNK_INDEX;
 134         lastChunkHashCode = INITIAL_LAST_CHUNK_HASH_CODE;
 135     }
 136
 137     int getLastChunkHashCode() {
 138         return lastChunkHashCode;
 139     }
 140 }