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 final int snapshotChunkSize;
  31     private final String logName;
  32     private ByteString snapshotBytes;
  33     private int offset = 0;
  34     // the next snapshot chunk is sent only if the replyReceivedForOffset matches offset
  35     private int replyReceivedForOffset = -1;
  36     // if replyStatus is false, the previous chunk is attempted
  37     private boolean replyStatus = false;
  38     private int chunkIndex = FIRST_CHUNK_INDEX;
  39     private int totalChunks;
  40     private int lastChunkHashCode = INITIAL_LAST_CHUNK_HASH_CODE;
  41     private int nextChunkHashCode = INITIAL_LAST_CHUNK_HASH_CODE;
  42
  43     LeaderInstallSnapshotState(int snapshotChunkSize, String logName) {
  44         this.snapshotChunkSize = snapshotChunkSize;
  45         this.logName = logName;
  46     }
  47
  48     ByteString getSnapshotBytes() {
  49         return snapshotBytes;
  50     }
  51
  52     void setSnapshotBytes(ByteString snapshotBytes) {
  53         if (this.snapshotBytes != null) {
  54             return;
  55         }
  56
  57         this.snapshotBytes = snapshotBytes;
  58         int size = snapshotBytes.size();
  59         totalChunks = size / snapshotChunkSize + (size % snapshotChunkSize > 0 ? 1 : 0);
  60
  61         LOG.debug("{}: Snapshot {} bytes, total chunks to send: {}", logName, size, totalChunks);
  62
  63         replyReceivedForOffset = -1;
  64         chunkIndex = FIRST_CHUNK_INDEX;
  65     }
  66
  67     int incrementOffset() {
  68         if (replyStatus) {
  69             // if prev chunk failed, we would want to sent the same chunk again
  70             offset = offset + snapshotChunkSize;
  71         }
  72         return offset;
  73     }
  74
  75     int incrementChunkIndex() {
  76         if (replyStatus) {
  77             // if prev chunk failed, we would want to sent the same chunk again
  78             chunkIndex =  chunkIndex + 1;
  79         }
  80         return chunkIndex;
  81     }
  82
  83     int getChunkIndex() {
  84         return chunkIndex;
  85     }
  86
  87     int getTotalChunks() {
  88         return totalChunks;
  89     }
  90
  91     boolean canSendNextChunk() {
  92         // we only send a false if a chunk is sent but we have not received a reply yet
  93         return snapshotBytes != null && replyReceivedForOffset == offset;
  94     }
  95
  96     boolean isLastChunk(int index) {
  97         return totalChunks == index;
  98     }
  99
 100     void markSendStatus(boolean success) {
 101         if (success) {
 102             // if the chunk sent was successful
 103             replyReceivedForOffset = offset;
 104             replyStatus = true;
 105             lastChunkHashCode = nextChunkHashCode;
 106         } else {
 107             // if the chunk sent was failure
 108             replyReceivedForOffset = offset;
 109             replyStatus = false;
 110         }
 111     }
 112
 113     byte[] getNextChunk() {
 114         int snapshotLength = getSnapshotBytes().size();
 115         int start = incrementOffset();
 116         int size = snapshotChunkSize;
 117         if (snapshotChunkSize > snapshotLength) {
 118             size = snapshotLength;
 119         } else if (start + snapshotChunkSize > snapshotLength) {
 120             size = snapshotLength - start;
 121         }
 122
 123         byte[] nextChunk = new byte[size];
 124         getSnapshotBytes().copyTo(nextChunk, start, 0, size);
 125         nextChunkHashCode = Arrays.hashCode(nextChunk);
 126
 127         LOG.debug("{}: Next chunk: total length={}, offset={}, size={}, hashCode={}", logName,
 128                 snapshotLength, start, size, nextChunkHashCode);
 129         return nextChunk;
 130     }
 131
 132     /**
 133      * Reset should be called when the Follower needs to be sent the snapshot from the beginning.
 134      */
 135     void reset() {
 136         offset = 0;
 137         replyStatus = false;
 138         replyReceivedForOffset = offset;
 139         chunkIndex = FIRST_CHUNK_INDEX;
 140         lastChunkHashCode = INITIAL_LAST_CHUNK_HASH_CODE;
 141     }
 142
 143     int getLastChunkHashCode() {
 144         return lastChunkHashCode;
 145     }
 146 }