opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/entityownership/EntityOwnershipShardCommitCoordinator.java

   1 /*
   2  * Copyright (c) 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.datastore.entityownership;
   9
  10 import static org.opendaylight.controller.cluster.datastore.entityownership.EntityOwnersModel.ENTITY_OWNER_QNAME;
  11
  12 import akka.actor.ActorRef;
  13 import akka.actor.Cancellable;
  14 import akka.actor.Status.Failure;
  15 import com.google.common.base.Preconditions;
  16 import java.util.Iterator;
  17 import java.util.LinkedList;
  18 import java.util.Queue;
  19 import javax.annotation.Nullable;
  20 import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier;
  21 import org.opendaylight.controller.cluster.access.concepts.FrontendIdentifier;
  22 import org.opendaylight.controller.cluster.access.concepts.FrontendType;
  23 import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier;
  24 import org.opendaylight.controller.cluster.access.concepts.MemberName;
  25 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
  26 import org.opendaylight.controller.cluster.datastore.DataStoreVersions;
  27 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
  28 import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
  29 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
  30 import org.opendaylight.controller.cluster.datastore.modification.Modification;
  31 import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
  32 import org.slf4j.Logger;
  33 import scala.concurrent.duration.FiniteDuration;
  34
  35 /**
  36  * Handles commits and retries for the EntityOwnershipShard.
  37  *
  38  * @author Thomas Pantelis
  39  */
  40 class EntityOwnershipShardCommitCoordinator {
  41     private static final Object COMMIT_RETRY_MESSAGE = new Object() {
  42         @Override
  43         public String toString() {
  44             return "entityCommitRetry";
  45         }
  46     };
  47     private static final FrontendType FRONTEND_TYPE = FrontendType.forName("entity-ownership-internal");
  48
  49     private final Queue<Modification> pendingModifications = new LinkedList<>();
  50     private final LocalHistoryIdentifier historyId;
  51     private final Logger log;
  52
  53     private BatchedModifications inflightCommit;
  54     private Cancellable retryCommitSchedule;
  55     private long transactionIDCounter = 0;
  56
  57     EntityOwnershipShardCommitCoordinator(MemberName localMemberName, Logger log) {
  58         this.log = Preconditions.checkNotNull(log);
  59         historyId = new LocalHistoryIdentifier(
  60                 ClientIdentifier.create(FrontendIdentifier.create(localMemberName, FRONTEND_TYPE), 0), 0);
  61     }
  62
  63     boolean handleMessage(Object message, EntityOwnershipShard shard) {
  64         boolean handled = true;
  65         if (CommitTransactionReply.isSerializedType(message)) {
  66             // Successful reply from a local commit.
  67             inflightCommitSucceeded(shard);
  68         } else if (message instanceof akka.actor.Status.Failure) {
  69             // Failure reply from a local commit.
  70             inflightCommitFailure(((Failure) message).cause(), shard);
  71         } else if (COMMIT_RETRY_MESSAGE.equals(message)) {
  72             retryInflightCommit(shard);
  73         } else {
  74             handled = false;
  75         }
  76
  77         return handled;
  78     }
  79
  80     private void retryInflightCommit(EntityOwnershipShard shard) {
  81         // Shouldn't be null happen but verify anyway
  82         if (inflightCommit == null) {
  83             return;
  84         }
  85
  86         if (shard.hasLeader()) {
  87             log.debug("Retrying commit for BatchedModifications {}", inflightCommit.getTransactionId());
  88
  89             shard.tryCommitModifications(inflightCommit);
  90         } else {
  91             scheduleInflightCommitRetry(shard);
  92         }
  93     }
  94
  95     void inflightCommitFailure(Throwable cause, EntityOwnershipShard shard) {
  96         // This should've originated from a failed inflight commit but verify anyway
  97         if (inflightCommit == null) {
  98             return;
  99         }
 100
 101         log.debug("Inflight BatchedModifications {} commit failed", inflightCommit.getTransactionId(), cause);
 102
 103         if (!(cause instanceof NoShardLeaderException)) {
 104             // If the failure is other than NoShardLeaderException the commit may have been partially
 105             // processed so retry with a new transaction ID to be safe.
 106             newInflightCommitWithDifferentTransactionID();
 107         }
 108
 109         scheduleInflightCommitRetry(shard);
 110     }
 111
 112     private void scheduleInflightCommitRetry(EntityOwnershipShard shard) {
 113         FiniteDuration duration = shard.getDatastoreContext().getShardRaftConfig().getElectionTimeOutInterval();
 114
 115         log.debug("Scheduling retry for BatchedModifications commit {} in {}",
 116                 inflightCommit.getTransactionId(), duration);
 117
 118         retryCommitSchedule = shard.getContext().system().scheduler().scheduleOnce(duration, shard.getSelf(),
 119                 COMMIT_RETRY_MESSAGE, shard.getContext().dispatcher(), ActorRef.noSender());
 120     }
 121
 122     void inflightCommitSucceeded(EntityOwnershipShard shard) {
 123         // Shouldn't be null but verify anyway
 124         if (inflightCommit == null) {
 125             return;
 126         }
 127
 128         if (retryCommitSchedule != null) {
 129             retryCommitSchedule.cancel();
 130         }
 131
 132         log.debug("BatchedModifications commit {} succeeded", inflightCommit.getTransactionId());
 133
 134         inflightCommit = null;
 135         commitNextBatch(shard);
 136     }
 137
 138     void commitNextBatch(EntityOwnershipShard shard) {
 139         if (inflightCommit != null || pendingModifications.isEmpty() || !shard.hasLeader()) {
 140             return;
 141         }
 142
 143         inflightCommit = newBatchedModifications();
 144         Iterator<Modification> iter = pendingModifications.iterator();
 145         while (iter.hasNext()) {
 146             inflightCommit.addModification(iter.next());
 147             iter.remove();
 148             if (inflightCommit.getModifications().size()
 149                     >= shard.getDatastoreContext().getShardBatchedModificationCount()) {
 150                 break;
 151             }
 152         }
 153
 154         log.debug("Committing next BatchedModifications {}, size {}", inflightCommit.getTransactionId(),
 155                 inflightCommit.getModifications().size());
 156
 157         shard.tryCommitModifications(inflightCommit);
 158     }
 159
 160     void commitModification(Modification modification, EntityOwnershipShard shard) {
 161         BatchedModifications modifications = newBatchedModifications();
 162         modifications.addModification(modification);
 163         commitModifications(modifications, shard);
 164     }
 165
 166     void commitModifications(BatchedModifications modifications, EntityOwnershipShard shard) {
 167         if (modifications.getModifications().isEmpty()) {
 168             return;
 169         }
 170
 171         boolean hasLeader = shard.hasLeader();
 172         if (inflightCommit != null || !hasLeader) {
 173             if (log.isDebugEnabled()) {
 174                 log.debug("{} - adding modifications to pending",
 175                         inflightCommit != null ? "A commit is inflight" : "No shard leader");
 176             }
 177
 178             pendingModifications.addAll(modifications.getModifications());
 179         } else {
 180             inflightCommit = modifications;
 181             shard.tryCommitModifications(inflightCommit);
 182         }
 183     }
 184
 185     void onStateChanged(EntityOwnershipShard shard, boolean isLeader) {
 186         shard.possiblyRemoveAllInitialCandidates(shard.getLeader());
 187
 188         possiblyPrunePendingCommits(shard, isLeader);
 189
 190         if (!isLeader && inflightCommit != null) {
 191             // We're no longer the leader but we have an inflight local commit. This likely means we didn't get
 192             // consensus for the commit and switched to follower due to another node with a higher term. We
 193             // can't be sure if the commit was replicated to any node so we retry it here with a new
 194             // transaction ID.
 195             if (retryCommitSchedule != null) {
 196                 retryCommitSchedule.cancel();
 197             }
 198
 199             newInflightCommitWithDifferentTransactionID();
 200             retryInflightCommit(shard);
 201         } else {
 202             commitNextBatch(shard);
 203         }
 204     }
 205
 206     private void possiblyPrunePendingCommits(EntityOwnershipShard shard, boolean isLeader) {
 207         // If we were the leader and transitioned to follower, we'll try to forward pending commits to the new leader.
 208         // However certain commits, e.g. entity owner changes, should only be committed by a valid leader as the
 209         // criteria used to determine the commit may be stale. Since we're no longer a valid leader, we should not
 210         // forward such commits thus we prune the pending modifications. We still should forward local candidate change
 211         // commits.
 212         if (shard.hasLeader() && !isLeader) {
 213             // We may have already submitted a transaction for replication and commit. We don't need the base Shard to
 214             // forward it since we also have it stored in the inflightCommit and handle retries. So we just clear
 215             // pending transactions and drop them.
 216             shard.convertPendingTransactionsToMessages();
 217
 218             // Prune the inflightCommit.
 219             if (inflightCommit != null) {
 220                 inflightCommit = pruneModifications(inflightCommit);
 221             }
 222
 223             // Prune the subsequent pending modifications.
 224             Iterator<Modification> iter = pendingModifications.iterator();
 225             while (iter.hasNext()) {
 226                 Modification mod = iter.next();
 227                 if (!canForwardModificationToNewLeader(mod)) {
 228                     iter.remove();
 229                 }
 230             }
 231         }
 232     }
 233
 234     @Nullable
 235     private BatchedModifications pruneModifications(BatchedModifications toPrune) {
 236         BatchedModifications prunedModifications = new BatchedModifications(toPrune.getTransactionId(),
 237                 toPrune.getVersion());
 238         prunedModifications.setDoCommitOnReady(toPrune.isDoCommitOnReady());
 239         prunedModifications.setReady(toPrune.isReady());
 240         prunedModifications.setTotalMessagesSent(toPrune.getTotalMessagesSent());
 241         for (Modification mod: toPrune.getModifications()) {
 242             if (canForwardModificationToNewLeader(mod)) {
 243                 prunedModifications.addModification(mod);
 244             }
 245         }
 246
 247         return !prunedModifications.getModifications().isEmpty() ? prunedModifications : null;
 248     }
 249
 250     private boolean canForwardModificationToNewLeader(Modification mod) {
 251         // If this is a WRITE of entity owner we don't want to forward it to a new leader since the criteria used
 252         // to determine the new owner might be stale.
 253         if (mod instanceof WriteModification) {
 254             WriteModification writeMod = (WriteModification)mod;
 255             boolean canForward = !writeMod.getPath().getLastPathArgument().getNodeType().equals(ENTITY_OWNER_QNAME);
 256
 257             if (!canForward) {
 258                 log.debug("Not forwarding WRITE modification for {} to new leader", writeMod.getPath());
 259             }
 260
 261             return canForward;
 262         }
 263
 264         return true;
 265     }
 266
 267     private void newInflightCommitWithDifferentTransactionID() {
 268         BatchedModifications newBatchedModifications = newBatchedModifications();
 269         newBatchedModifications.addModifications(inflightCommit.getModifications());
 270         inflightCommit = newBatchedModifications;
 271     }
 272
 273     BatchedModifications newBatchedModifications() {
 274         BatchedModifications modifications = new BatchedModifications(
 275             new TransactionIdentifier(historyId, ++transactionIDCounter), DataStoreVersions.CURRENT_VERSION);
 276         modifications.setDoCommitOnReady(true);
 277         modifications.setReady(true);
 278         modifications.setTotalMessagesSent(1);
 279         return modifications;
 280     }
 281 }