opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.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
   9 package org.opendaylight.controller.cluster.datastore;
  10
  11 import akka.actor.ActorRef;
  12 import akka.actor.ActorSelection;
  13 import akka.actor.Cancellable;
  14 import akka.actor.Props;
  15 import akka.japi.Creator;
  16 import akka.persistence.RecoveryFailure;
  17 import akka.serialization.Serialization;
  18 import com.google.common.annotations.VisibleForTesting;
  19 import com.google.common.base.Optional;
  20 import com.google.common.base.Preconditions;
  21 import com.google.common.collect.Lists;
  22 import com.google.common.util.concurrent.FutureCallback;
  23 import com.google.common.util.concurrent.Futures;
  24 import com.google.common.util.concurrent.ListenableFuture;
  25 import java.io.IOException;
  26 import java.util.Collection;
  27 import java.util.HashMap;
  28 import java.util.List;
  29 import java.util.Map;
  30 import java.util.concurrent.ExecutionException;
  31 import java.util.concurrent.TimeUnit;
  32 import javax.annotation.Nonnull;
  33 import org.opendaylight.controller.cluster.DataPersistenceProvider;
  34 import org.opendaylight.controller.cluster.common.actor.CommonConfig;
  35 import org.opendaylight.controller.cluster.common.actor.MeteringBehavior;
  36 import org.opendaylight.controller.cluster.datastore.ShardCommitCoordinator.CohortEntry;
  37 import org.opendaylight.controller.cluster.datastore.compat.BackwardsCompatibleThreePhaseCommitCohort;
  38 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
  39 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
  40 import org.opendaylight.controller.cluster.datastore.identifiers.ShardTransactionIdentifier;
  41 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardMBeanFactory;
  42 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
  43 import org.opendaylight.controller.cluster.datastore.messages.AbortTransaction;
  44 import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply;
  45 import org.opendaylight.controller.cluster.datastore.messages.ActorInitialized;
  46 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction;
  47 import org.opendaylight.controller.cluster.datastore.messages.CloseTransactionChain;
  48 import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction;
  49 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
  50 import org.opendaylight.controller.cluster.datastore.messages.CreateSnapshot;
  51 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
  52 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
  53 import org.opendaylight.controller.cluster.datastore.messages.EnableNotification;
  54 import org.opendaylight.controller.cluster.datastore.messages.ForwardedReadyTransaction;
  55 import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved;
  56 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
  57 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
  58 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
  59 import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
  60 import org.opendaylight.controller.cluster.datastore.modification.Modification;
  61 import org.opendaylight.controller.cluster.datastore.modification.ModificationPayload;
  62 import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
  63 import org.opendaylight.controller.cluster.datastore.utils.SerializationUtils;
  64 import org.opendaylight.controller.cluster.notifications.RoleChangeNotifier;
  65 import org.opendaylight.controller.cluster.raft.RaftActor;
  66 import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
  67 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.CompositeModificationByteStringPayload;
  68 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.CompositeModificationPayload;
  69 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
  70 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
  71 import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore;
  72 import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStoreFactory;
  73 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
  74 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
  75 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionFactory;
  76 import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
  77 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  78 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  79 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  80 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  81 import scala.concurrent.duration.Duration;
  82 import scala.concurrent.duration.FiniteDuration;
  83
  84 /**
  85  * A Shard represents a portion of the logical data tree <br/>
  86  * <p>
  87  * Our Shard uses InMemoryDataStore as it's internal representation and delegates all requests it
  88  * </p>
  89  */
  90 public class Shard extends RaftActor {
  91
  92     private static final YangInstanceIdentifier DATASTORE_ROOT = YangInstanceIdentifier.builder().build();
  93
  94     private static final Object TX_COMMIT_TIMEOUT_CHECK_MESSAGE = "txCommitTimeoutCheck";
  95
  96     @VisibleForTesting
  97     static final String DEFAULT_NAME = "default";
  98
  99     // The state of this Shard
 100     private final InMemoryDOMDataStore store;
 101
 102     /// The name of this shard
 103     private final ShardIdentifier name;
 104
 105     private final ShardStats shardMBean;
 106
 107     private final List<ActorSelection> dataChangeListeners =  Lists.newArrayList();
 108
 109     private final List<DelayedListenerRegistration> delayedListenerRegistrations =
 110                                                                        Lists.newArrayList();
 111
 112     private final DatastoreContext datastoreContext;
 113
 114     private final DataPersistenceProvider dataPersistenceProvider;
 115
 116     private SchemaContext schemaContext;
 117
 118     private int createSnapshotTransactionCounter;
 119
 120     private final ShardCommitCoordinator commitCoordinator;
 121
 122     private final long transactionCommitTimeout;
 123
 124     private Cancellable txCommitTimeoutCheckSchedule;
 125
 126     private final Optional<ActorRef> roleChangeNotifier;
 127
 128     /**
 129      * Coordinates persistence recovery on startup.
 130      */
 131     private ShardRecoveryCoordinator recoveryCoordinator;
 132     private List<Object> currentLogRecoveryBatch;
 133
 134     private final Map<String, DOMStoreTransactionChain> transactionChains = new HashMap<>();
 135
 136     protected Shard(final ShardIdentifier name, final Map<ShardIdentifier, String> peerAddresses,
 137             final DatastoreContext datastoreContext, final SchemaContext schemaContext) {
 138         super(name.toString(), mapPeerAddresses(peerAddresses),
 139                 Optional.of(datastoreContext.getShardRaftConfig()));
 140
 141         this.name = name;
 142         this.datastoreContext = datastoreContext;
 143         this.schemaContext = schemaContext;
 144         this.dataPersistenceProvider = (datastoreContext.isPersistent()) ? new PersistentDataProvider() : new NonPersistentRaftDataProvider();
 145
 146         LOG.info("Shard created : {}, persistent : {}", name, datastoreContext.isPersistent());
 147
 148         store = InMemoryDOMDataStoreFactory.create(name.toString(), null,
 149                 datastoreContext.getDataStoreProperties());
 150
 151         if(schemaContext != null) {
 152             store.onGlobalContextUpdated(schemaContext);
 153         }
 154
 155         shardMBean = ShardMBeanFactory.getShardStatsMBean(name.toString(),
 156                 datastoreContext.getDataStoreMXBeanType());
 157         shardMBean.setNotificationManager(store.getDataChangeListenerNotificationManager());
 158
 159         if (isMetricsCaptureEnabled()) {
 160             getContext().become(new MeteringBehavior(this));
 161         }
 162
 163         commitCoordinator = new ShardCommitCoordinator(TimeUnit.SECONDS.convert(1, TimeUnit.MINUTES),
 164                 datastoreContext.getShardTransactionCommitQueueCapacity(), LOG, name.toString());
 165
 166         transactionCommitTimeout = TimeUnit.MILLISECONDS.convert(
 167                 datastoreContext.getShardTransactionCommitTimeoutInSeconds(), TimeUnit.SECONDS);
 168
 169         // create a notifier actor for each cluster member
 170         roleChangeNotifier = createRoleChangeNotifier(name.toString());
 171     }
 172
 173     private static Map<String, String> mapPeerAddresses(
 174         final Map<ShardIdentifier, String> peerAddresses) {
 175         Map<String, String> map = new HashMap<>();
 176
 177         for (Map.Entry<ShardIdentifier, String> entry : peerAddresses
 178             .entrySet()) {
 179             map.put(entry.getKey().toString(), entry.getValue());
 180         }
 181
 182         return map;
 183     }
 184
 185     public static Props props(final ShardIdentifier name,
 186         final Map<ShardIdentifier, String> peerAddresses,
 187         final DatastoreContext datastoreContext, final SchemaContext schemaContext) {
 188         Preconditions.checkNotNull(name, "name should not be null");
 189         Preconditions.checkNotNull(peerAddresses, "peerAddresses should not be null");
 190         Preconditions.checkNotNull(datastoreContext, "dataStoreContext should not be null");
 191         Preconditions.checkNotNull(schemaContext, "schemaContext should not be null");
 192
 193         return Props.create(new ShardCreator(name, peerAddresses, datastoreContext, schemaContext));
 194     }
 195
 196     private Optional<ActorRef> createRoleChangeNotifier(String shardId) {
 197         ActorRef shardRoleChangeNotifier = this.getContext().actorOf(
 198             RoleChangeNotifier.getProps(shardId), shardId + "-notifier");
 199         return Optional.<ActorRef>of(shardRoleChangeNotifier);
 200     }
 201
 202     @Override
 203     public void postStop() {
 204         super.postStop();
 205
 206         if(txCommitTimeoutCheckSchedule != null) {
 207             txCommitTimeoutCheckSchedule.cancel();
 208         }
 209     }
 210
 211     @Override
 212     public void onReceiveRecover(final Object message) throws Exception {
 213         if(LOG.isDebugEnabled()) {
 214             LOG.debug("{}: onReceiveRecover: Received message {} from {}", persistenceId(),
 215                 message.getClass().toString(), getSender());
 216         }
 217
 218         if (message instanceof RecoveryFailure){
 219             LOG.error("{}: Recovery failed because of this cause",
 220                     persistenceId(), ((RecoveryFailure) message).cause());
 221
 222             // Even though recovery failed, we still need to finish our recovery, eg send the
 223             // ActorInitialized message and start the txCommitTimeoutCheckSchedule.
 224             onRecoveryComplete();
 225         } else {
 226             super.onReceiveRecover(message);
 227         }
 228     }
 229
 230     @Override
 231     public void onReceiveCommand(final Object message) throws Exception {
 232         if (message.getClass().equals(CreateTransaction.SERIALIZABLE_CLASS)) {
 233             handleCreateTransaction(message);
 234         } else if(message instanceof ForwardedReadyTransaction) {
 235             handleForwardedReadyTransaction((ForwardedReadyTransaction)message);
 236         } else if(message.getClass().equals(CanCommitTransaction.SERIALIZABLE_CLASS)) {
 237             handleCanCommitTransaction(CanCommitTransaction.fromSerializable(message));
 238         } else if(message.getClass().equals(CommitTransaction.SERIALIZABLE_CLASS)) {
 239             handleCommitTransaction(CommitTransaction.fromSerializable(message));
 240         } else if(message.getClass().equals(AbortTransaction.SERIALIZABLE_CLASS)) {
 241             handleAbortTransaction(AbortTransaction.fromSerializable(message));
 242         } else if (message.getClass().equals(CloseTransactionChain.SERIALIZABLE_CLASS)){
 243             closeTransactionChain(CloseTransactionChain.fromSerializable(message));
 244         } else if (message instanceof RegisterChangeListener) {
 245             registerChangeListener((RegisterChangeListener) message);
 246         } else if (message instanceof UpdateSchemaContext) {
 247             updateSchemaContext((UpdateSchemaContext) message);
 248         } else if (message instanceof PeerAddressResolved) {
 249             PeerAddressResolved resolved = (PeerAddressResolved) message;
 250             setPeerAddress(resolved.getPeerId().toString(),
 251                 resolved.getPeerAddress());
 252         } else if(message.equals(TX_COMMIT_TIMEOUT_CHECK_MESSAGE)) {
 253             handleTransactionCommitTimeoutCheck();
 254         } else {
 255             super.onReceiveCommand(message);
 256         }
 257     }
 258
 259     @Override
 260     protected Optional<ActorRef> getRoleChangeNotifier() {
 261         return roleChangeNotifier;
 262     }
 263
 264     private void handleTransactionCommitTimeoutCheck() {
 265         CohortEntry cohortEntry = commitCoordinator.getCurrentCohortEntry();
 266         if(cohortEntry != null) {
 267             long elapsed = System.currentTimeMillis() - cohortEntry.getLastAccessTime();
 268             if(elapsed > transactionCommitTimeout) {
 269                 LOG.warn("{}: Current transaction {} has timed out after {} ms - aborting",
 270                         persistenceId(), cohortEntry.getTransactionID(), transactionCommitTimeout);
 271
 272                 doAbortTransaction(cohortEntry.getTransactionID(), null);
 273             }
 274         }
 275     }
 276
 277     private void handleCommitTransaction(final CommitTransaction commit) {
 278         final String transactionID = commit.getTransactionID();
 279
 280         LOG.debug("{}: Committing transaction {}", persistenceId(), transactionID);
 281
 282         // Get the current in-progress cohort entry in the commitCoordinator if it corresponds to
 283         // this transaction.
 284         final CohortEntry cohortEntry = commitCoordinator.getCohortEntryIfCurrent(transactionID);
 285         if(cohortEntry == null) {
 286             // We're not the current Tx - the Tx was likely expired b/c it took too long in
 287             // between the canCommit and commit messages.
 288             IllegalStateException ex = new IllegalStateException(
 289                     String.format("%s: Cannot commit transaction %s - it is not the current transaction",
 290                             persistenceId(), transactionID));
 291             LOG.error(ex.getMessage());
 292             shardMBean.incrementFailedTransactionsCount();
 293             getSender().tell(new akka.actor.Status.Failure(ex), getSelf());
 294             return;
 295         }
 296
 297         // We perform the preCommit phase here atomically with the commit phase. This is an
 298         // optimization to eliminate the overhead of an extra preCommit message. We lose front-end
 299         // coordination of preCommit across shards in case of failure but preCommit should not
 300         // normally fail since we ensure only one concurrent 3-phase commit.
 301
 302         try {
 303             // We block on the future here so we don't have to worry about possibly accessing our
 304             // state on a different thread outside of our dispatcher. Also, the data store
 305             // currently uses a same thread executor anyway.
 306             cohortEntry.getCohort().preCommit().get();
 307
 308             // If we do not have any followers and we are not using persistence we can
 309             // apply modification to the state immediately
 310             if(!hasFollowers() && !persistence().isRecoveryApplicable()){
 311                 applyModificationToState(getSender(), transactionID, cohortEntry.getModification());
 312             } else {
 313                 Shard.this.persistData(getSender(), transactionID,
 314                         new ModificationPayload(cohortEntry.getModification()));
 315             }
 316         } catch (Exception e) {
 317             LOG.error("{} An exception occurred while preCommitting transaction {}",
 318                     persistenceId(), cohortEntry.getTransactionID(), e);
 319             shardMBean.incrementFailedTransactionsCount();
 320             getSender().tell(new akka.actor.Status.Failure(e), getSelf());
 321         }
 322
 323         cohortEntry.updateLastAccessTime();
 324     }
 325
 326     private void finishCommit(@Nonnull final ActorRef sender, final @Nonnull String transactionID) {
 327         // With persistence enabled, this method is called via applyState by the leader strategy
 328         // after the commit has been replicated to a majority of the followers.
 329
 330         CohortEntry cohortEntry = commitCoordinator.getCohortEntryIfCurrent(transactionID);
 331         if(cohortEntry == null) {
 332             // The transaction is no longer the current commit. This can happen if the transaction
 333             // was aborted prior, most likely due to timeout in the front-end. We need to finish
 334             // committing the transaction though since it was successfully persisted and replicated
 335             // however we can't use the original cohort b/c it was already preCommitted and may
 336             // conflict with the current commit or may have been aborted so we commit with a new
 337             // transaction.
 338             cohortEntry = commitCoordinator.getAndRemoveCohortEntry(transactionID);
 339             if(cohortEntry != null) {
 340                 commitWithNewTransaction(cohortEntry.getModification());
 341                 sender.tell(CommitTransactionReply.INSTANCE.toSerializable(), getSelf());
 342             } else {
 343                 // This really shouldn't happen - it likely means that persistence or replication
 344                 // took so long to complete such that the cohort entry was expired from the cache.
 345                 IllegalStateException ex = new IllegalStateException(
 346                         String.format("%s: Could not finish committing transaction %s - no CohortEntry found",
 347                                 persistenceId(), transactionID));
 348                 LOG.error(ex.getMessage());
 349                 sender.tell(new akka.actor.Status.Failure(ex), getSelf());
 350             }
 351
 352             return;
 353         }
 354
 355         LOG.debug("{}: Finishing commit for transaction {}", persistenceId(), cohortEntry.getTransactionID());
 356
 357         try {
 358             // We block on the future here so we don't have to worry about possibly accessing our
 359             // state on a different thread outside of our dispatcher. Also, the data store
 360             // currently uses a same thread executor anyway.
 361             cohortEntry.getCohort().commit().get();
 362
 363             sender.tell(CommitTransactionReply.INSTANCE.toSerializable(), getSelf());
 364
 365             shardMBean.incrementCommittedTransactionCount();
 366             shardMBean.setLastCommittedTransactionTime(System.currentTimeMillis());
 367
 368         } catch (Exception e) {
 369             sender.tell(new akka.actor.Status.Failure(e), getSelf());
 370
 371             LOG.error("{}, An exception occurred while committing transaction {}", persistenceId(),
 372                     transactionID, e);
 373             shardMBean.incrementFailedTransactionsCount();
 374         } finally {
 375             commitCoordinator.currentTransactionComplete(transactionID, true);
 376         }
 377     }
 378
 379     private void handleCanCommitTransaction(final CanCommitTransaction canCommit) {
 380         LOG.debug("{}: Can committing transaction {}", persistenceId(), canCommit.getTransactionID());
 381         commitCoordinator.handleCanCommit(canCommit, getSender(), self());
 382     }
 383
 384     private void handleForwardedReadyTransaction(ForwardedReadyTransaction ready) {
 385         LOG.debug("{}: Readying transaction {}, client version {}", persistenceId(),
 386                 ready.getTransactionID(), ready.getTxnClientVersion());
 387
 388         // This message is forwarded by the ShardTransaction on ready. We cache the cohort in the
 389         // commitCoordinator in preparation for the subsequent three phase commit initiated by
 390         // the front-end.
 391         commitCoordinator.transactionReady(ready.getTransactionID(), ready.getCohort(),
 392                 ready.getModification());
 393
 394         // Return our actor path as we'll handle the three phase commit, except if the Tx client
 395         // version < 1 (Helium-1 version). This means the Tx was initiated by a base Helium version
 396         // node. In that case, the subsequent 3-phase commit messages won't contain the
 397         // transactionId so to maintain backwards compatibility, we create a separate cohort actor
 398         // to provide the compatible behavior.
 399         ActorRef replyActorPath = self();
 400         if(ready.getTxnClientVersion() < DataStoreVersions.HELIUM_1_VERSION) {
 401             LOG.debug("{}: Creating BackwardsCompatibleThreePhaseCommitCohort", persistenceId());
 402             replyActorPath = getContext().actorOf(BackwardsCompatibleThreePhaseCommitCohort.props(
 403                     ready.getTransactionID()));
 404         }
 405
 406         ReadyTransactionReply readyTransactionReply = new ReadyTransactionReply(
 407                 Serialization.serializedActorPath(replyActorPath));
 408         getSender().tell(ready.isReturnSerialized() ? readyTransactionReply.toSerializable() :
 409                 readyTransactionReply, getSelf());
 410     }
 411
 412     private void handleAbortTransaction(final AbortTransaction abort) {
 413         doAbortTransaction(abort.getTransactionID(), getSender());
 414     }
 415
 416     void doAbortTransaction(final String transactionID, final ActorRef sender) {
 417         final CohortEntry cohortEntry = commitCoordinator.getCohortEntryIfCurrent(transactionID);
 418         if(cohortEntry != null) {
 419             LOG.debug("{}: Aborting transaction {}", persistenceId(), transactionID);
 420
 421             // We don't remove the cached cohort entry here (ie pass false) in case the Tx was
 422             // aborted during replication in which case we may still commit locally if replication
 423             // succeeds.
 424             commitCoordinator.currentTransactionComplete(transactionID, false);
 425
 426             final ListenableFuture<Void> future = cohortEntry.getCohort().abort();
 427             final ActorRef self = getSelf();
 428
 429             Futures.addCallback(future, new FutureCallback<Void>() {
 430                 @Override
 431                 public void onSuccess(final Void v) {
 432                     shardMBean.incrementAbortTransactionsCount();
 433
 434                     if(sender != null) {
 435                         sender.tell(AbortTransactionReply.INSTANCE.toSerializable(), self);
 436                     }
 437                 }
 438
 439                 @Override
 440                 public void onFailure(final Throwable t) {
 441                     LOG.error("{}: An exception happened during abort", persistenceId(), t);
 442
 443                     if(sender != null) {
 444                         sender.tell(new akka.actor.Status.Failure(t), self);
 445                     }
 446                 }
 447             });
 448         }
 449     }
 450
 451     private void handleCreateTransaction(final Object message) {
 452         if (isLeader()) {
 453             createTransaction(CreateTransaction.fromSerializable(message));
 454         } else if (getLeader() != null) {
 455             getLeader().forward(message, getContext());
 456         } else {
 457             getSender().tell(new akka.actor.Status.Failure(new NoShardLeaderException(String.format(
 458                 "Could not find leader for shard %s so transaction cannot be created. This typically happens" +
 459                 " when the system is coming up or recovering and a leader is being elected. Try again" +
 460                 " later.", persistenceId()))), getSelf());
 461         }
 462     }
 463
 464     private void closeTransactionChain(final CloseTransactionChain closeTransactionChain) {
 465         DOMStoreTransactionChain chain =
 466             transactionChains.remove(closeTransactionChain.getTransactionChainId());
 467
 468         if(chain != null) {
 469             chain.close();
 470         }
 471     }
 472
 473     private ActorRef createTypedTransactionActor(int transactionType,
 474             ShardTransactionIdentifier transactionId, String transactionChainId,
 475             short clientVersion ) {
 476
 477         DOMStoreTransactionFactory factory = store;
 478
 479         if(!transactionChainId.isEmpty()) {
 480             factory = transactionChains.get(transactionChainId);
 481             if(factory == null){
 482                 DOMStoreTransactionChain transactionChain = store.createTransactionChain();
 483                 transactionChains.put(transactionChainId, transactionChain);
 484                 factory = transactionChain;
 485             }
 486         }
 487
 488         if(this.schemaContext == null) {
 489             throw new IllegalStateException("SchemaContext is not set");
 490         }
 491
 492         if (transactionType == TransactionProxy.TransactionType.READ_ONLY.ordinal()) {
 493
 494             shardMBean.incrementReadOnlyTransactionCount();
 495
 496             return getContext().actorOf(
 497                 ShardTransaction.props(factory.newReadOnlyTransaction(), getSelf(),
 498                         schemaContext,datastoreContext, shardMBean,
 499                         transactionId.getRemoteTransactionId(), clientVersion),
 500                         transactionId.toString());
 501
 502         } else if (transactionType == TransactionProxy.TransactionType.READ_WRITE.ordinal()) {
 503
 504             shardMBean.incrementReadWriteTransactionCount();
 505
 506             return getContext().actorOf(
 507                 ShardTransaction.props(factory.newReadWriteTransaction(), getSelf(),
 508                         schemaContext, datastoreContext, shardMBean,
 509                         transactionId.getRemoteTransactionId(), clientVersion),
 510                         transactionId.toString());
 511
 512
 513         } else if (transactionType == TransactionProxy.TransactionType.WRITE_ONLY.ordinal()) {
 514
 515             shardMBean.incrementWriteOnlyTransactionCount();
 516
 517             return getContext().actorOf(
 518                 ShardTransaction.props(factory.newWriteOnlyTransaction(), getSelf(),
 519                         schemaContext, datastoreContext, shardMBean,
 520                         transactionId.getRemoteTransactionId(), clientVersion),
 521                         transactionId.toString());
 522         } else {
 523             throw new IllegalArgumentException(
 524                 "Shard="+name + ":CreateTransaction message has unidentified transaction type="
 525                     + transactionType);
 526         }
 527     }
 528
 529     private void createTransaction(CreateTransaction createTransaction) {
 530         try {
 531             ActorRef transactionActor = createTransaction(createTransaction.getTransactionType(),
 532                 createTransaction.getTransactionId(), createTransaction.getTransactionChainId(),
 533                 createTransaction.getVersion());
 534
 535             getSender().tell(new CreateTransactionReply(Serialization.serializedActorPath(transactionActor),
 536                     createTransaction.getTransactionId()).toSerializable(), getSelf());
 537         } catch (Exception e) {
 538             getSender().tell(new akka.actor.Status.Failure(e), getSelf());
 539         }
 540     }
 541
 542     private ActorRef createTransaction(int transactionType, String remoteTransactionId,
 543             String transactionChainId, short clientVersion) {
 544
 545         ShardTransactionIdentifier transactionId =
 546             ShardTransactionIdentifier.builder()
 547                 .remoteTransactionId(remoteTransactionId)
 548                 .build();
 549
 550         if(LOG.isDebugEnabled()) {
 551             LOG.debug("{}: Creating transaction : {} ", persistenceId(), transactionId);
 552         }
 553
 554         ActorRef transactionActor = createTypedTransactionActor(transactionType, transactionId,
 555                 transactionChainId, clientVersion);
 556
 557         return transactionActor;
 558     }
 559
 560     private void syncCommitTransaction(final DOMStoreWriteTransaction transaction)
 561         throws ExecutionException, InterruptedException {
 562         DOMStoreThreePhaseCommitCohort commitCohort = transaction.ready();
 563         commitCohort.preCommit().get();
 564         commitCohort.commit().get();
 565     }
 566
 567     private void commitWithNewTransaction(final Modification modification) {
 568         DOMStoreWriteTransaction tx = store.newWriteOnlyTransaction();
 569         modification.apply(tx);
 570         try {
 571             syncCommitTransaction(tx);
 572             shardMBean.incrementCommittedTransactionCount();
 573             shardMBean.setLastCommittedTransactionTime(System.currentTimeMillis());
 574         } catch (InterruptedException | ExecutionException e) {
 575             shardMBean.incrementFailedTransactionsCount();
 576             LOG.error("{}: Failed to commit", persistenceId(), e);
 577         }
 578     }
 579
 580     private void updateSchemaContext(final UpdateSchemaContext message) {
 581         this.schemaContext = message.getSchemaContext();
 582         updateSchemaContext(message.getSchemaContext());
 583         store.onGlobalContextUpdated(message.getSchemaContext());
 584     }
 585
 586     @VisibleForTesting
 587     void updateSchemaContext(final SchemaContext schemaContext) {
 588         store.onGlobalContextUpdated(schemaContext);
 589     }
 590
 591     private void registerChangeListener(final RegisterChangeListener registerChangeListener) {
 592
 593         LOG.debug("{}: registerDataChangeListener for {}", persistenceId(), registerChangeListener.getPath());
 594
 595         ListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier,
 596                                                      NormalizedNode<?, ?>>> registration;
 597         if(isLeader()) {
 598             registration = doChangeListenerRegistration(registerChangeListener);
 599         } else {
 600             LOG.debug("{}: Shard is not the leader - delaying registration", persistenceId());
 601
 602             DelayedListenerRegistration delayedReg =
 603                     new DelayedListenerRegistration(registerChangeListener);
 604             delayedListenerRegistrations.add(delayedReg);
 605             registration = delayedReg;
 606         }
 607
 608         ActorRef listenerRegistration = getContext().actorOf(
 609                 DataChangeListenerRegistration.props(registration));
 610
 611         LOG.debug("{}: registerDataChangeListener sending reply, listenerRegistrationPath = {} ",
 612                 persistenceId(), listenerRegistration.path());
 613
 614         getSender().tell(new RegisterChangeListenerReply(listenerRegistration.path()), getSelf());
 615     }
 616
 617     private ListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier,
 618                                                NormalizedNode<?, ?>>> doChangeListenerRegistration(
 619             final RegisterChangeListener registerChangeListener) {
 620
 621         ActorSelection dataChangeListenerPath = getContext().system().actorSelection(
 622                 registerChangeListener.getDataChangeListenerPath());
 623
 624         // Notify the listener if notifications should be enabled or not
 625         // If this shard is the leader then it will enable notifications else
 626         // it will not
 627         dataChangeListenerPath.tell(new EnableNotification(true), getSelf());
 628
 629         // Now store a reference to the data change listener so it can be notified
 630         // at a later point if notifications should be enabled or disabled
 631         dataChangeListeners.add(dataChangeListenerPath);
 632
 633         AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>> listener =
 634                 new DataChangeListenerProxy(dataChangeListenerPath);
 635
 636         LOG.debug("{}: Registering for path {}", persistenceId(), registerChangeListener.getPath());
 637
 638         return store.registerChangeListener(registerChangeListener.getPath(), listener,
 639                 registerChangeListener.getScope());
 640     }
 641
 642     private boolean isMetricsCaptureEnabled(){
 643         CommonConfig config = new CommonConfig(getContext().system().settings().config());
 644         return config.isMetricCaptureEnabled();
 645     }
 646
 647     @Override
 648     protected
 649     void startLogRecoveryBatch(final int maxBatchSize) {
 650         currentLogRecoveryBatch = Lists.newArrayListWithCapacity(maxBatchSize);
 651
 652         if(LOG.isDebugEnabled()) {
 653             LOG.debug("{}: starting log recovery batch with max size {}", persistenceId(), maxBatchSize);
 654         }
 655     }
 656
 657     @Override
 658     protected void appendRecoveredLogEntry(final Payload data) {
 659         if(data instanceof ModificationPayload) {
 660             try {
 661                 currentLogRecoveryBatch.add(((ModificationPayload) data).getModification());
 662             } catch (ClassNotFoundException | IOException e) {
 663                 LOG.error("{}: Error extracting ModificationPayload", persistenceId(), e);
 664             }
 665         } else if (data instanceof CompositeModificationPayload) {
 666             currentLogRecoveryBatch.add(((CompositeModificationPayload) data).getModification());
 667         } else if (data instanceof CompositeModificationByteStringPayload) {
 668             currentLogRecoveryBatch.add(((CompositeModificationByteStringPayload) data).getModification());
 669         } else {
 670             LOG.error("{}: Unknown state received {} during recovery", persistenceId(), data);
 671         }
 672     }
 673
 674     @Override
 675     protected void applyRecoverySnapshot(final byte[] snapshotBytes) {
 676         if(recoveryCoordinator == null) {
 677             recoveryCoordinator = new ShardRecoveryCoordinator(persistenceId(), schemaContext,
 678                     LOG, name.toString());
 679         }
 680
 681         recoveryCoordinator.submit(snapshotBytes, store.newWriteOnlyTransaction());
 682
 683         if(LOG.isDebugEnabled()) {
 684             LOG.debug("{}: submitted recovery sbapshot", persistenceId());
 685         }
 686     }
 687
 688     @Override
 689     protected void applyCurrentLogRecoveryBatch() {
 690         if(recoveryCoordinator == null) {
 691             recoveryCoordinator = new ShardRecoveryCoordinator(persistenceId(), schemaContext,
 692                     LOG, name.toString());
 693         }
 694
 695         recoveryCoordinator.submit(currentLogRecoveryBatch, store.newWriteOnlyTransaction());
 696
 697         if(LOG.isDebugEnabled()) {
 698             LOG.debug("{}: submitted log recovery batch with size {}", persistenceId(),
 699                     currentLogRecoveryBatch.size());
 700         }
 701     }
 702
 703     @Override
 704     protected void onRecoveryComplete() {
 705         if(recoveryCoordinator != null) {
 706             Collection<DOMStoreWriteTransaction> txList = recoveryCoordinator.getTransactions();
 707
 708             if(LOG.isDebugEnabled()) {
 709                 LOG.debug("{}: recovery complete - committing {} Tx's", persistenceId(), txList.size());
 710             }
 711
 712             for(DOMStoreWriteTransaction tx: txList) {
 713                 try {
 714                     syncCommitTransaction(tx);
 715                     shardMBean.incrementCommittedTransactionCount();
 716                 } catch (InterruptedException | ExecutionException e) {
 717                     shardMBean.incrementFailedTransactionsCount();
 718                     LOG.error("{}: Failed to commit", persistenceId(), e);
 719                 }
 720             }
 721         }
 722
 723         recoveryCoordinator = null;
 724         currentLogRecoveryBatch = null;
 725         updateJournalStats();
 726
 727         //notify shard manager
 728         getContext().parent().tell(new ActorInitialized(), getSelf());
 729
 730         // Being paranoid here - this method should only be called once but just in case...
 731         if(txCommitTimeoutCheckSchedule == null) {
 732             // Schedule a message to be periodically sent to check if the current in-progress
 733             // transaction should be expired and aborted.
 734             FiniteDuration period = Duration.create(transactionCommitTimeout / 3, TimeUnit.MILLISECONDS);
 735             txCommitTimeoutCheckSchedule = getContext().system().scheduler().schedule(
 736                     period, period, getSelf(),
 737                     TX_COMMIT_TIMEOUT_CHECK_MESSAGE, getContext().dispatcher(), ActorRef.noSender());
 738         }
 739     }
 740
 741     @Override
 742     protected void applyState(final ActorRef clientActor, final String identifier, final Object data) {
 743
 744         if(data instanceof ModificationPayload) {
 745             try {
 746                 applyModificationToState(clientActor, identifier, ((ModificationPayload) data).getModification());
 747             } catch (ClassNotFoundException | IOException e) {
 748                 LOG.error("{}: Error extracting ModificationPayload", persistenceId(), e);
 749             }
 750         }
 751         else if (data instanceof CompositeModificationPayload) {
 752             Object modification = ((CompositeModificationPayload) data).getModification();
 753
 754             applyModificationToState(clientActor, identifier, modification);
 755         } else if(data instanceof CompositeModificationByteStringPayload ){
 756             Object modification = ((CompositeModificationByteStringPayload) data).getModification();
 757
 758             applyModificationToState(clientActor, identifier, modification);
 759         } else {
 760             LOG.error("{}: Unknown state received {} Class loader = {} CompositeNodeMod.ClassLoader = {}",
 761                     persistenceId(), data, data.getClass().getClassLoader(),
 762                     CompositeModificationPayload.class.getClassLoader());
 763         }
 764
 765         updateJournalStats();
 766
 767     }
 768
 769     private void applyModificationToState(ActorRef clientActor, String identifier, Object modification) {
 770         if(modification == null) {
 771             LOG.error(
 772                     "{}: modification is null - this is very unexpected, clientActor = {}, identifier = {}",
 773                     persistenceId(), identifier, clientActor != null ? clientActor.path().toString() : null);
 774         } else if(clientActor == null) {
 775             // There's no clientActor to which to send a commit reply so we must be applying
 776             // replicated state from the leader.
 777             commitWithNewTransaction(MutableCompositeModification.fromSerializable(modification));
 778         } else {
 779             // This must be the OK to commit after replication consensus.
 780             finishCommit(clientActor, identifier);
 781         }
 782     }
 783
 784     private void updateJournalStats() {
 785         ReplicatedLogEntry lastLogEntry = getLastLogEntry();
 786
 787         if (lastLogEntry != null) {
 788             shardMBean.setLastLogIndex(lastLogEntry.getIndex());
 789             shardMBean.setLastLogTerm(lastLogEntry.getTerm());
 790         }
 791
 792         shardMBean.setCommitIndex(getCommitIndex());
 793         shardMBean.setLastApplied(getLastApplied());
 794         shardMBean.setInMemoryJournalDataSize(getRaftActorContext().getReplicatedLog().dataSize());
 795     }
 796
 797     @Override
 798     protected void createSnapshot() {
 799         // Create a transaction actor. We are really going to treat the transaction as a worker
 800         // so that this actor does not get block building the snapshot. THe transaction actor will
 801         // after processing the CreateSnapshot message.
 802
 803         ActorRef createSnapshotTransaction = createTransaction(
 804                 TransactionProxy.TransactionType.READ_ONLY.ordinal(),
 805                 "createSnapshot" + ++createSnapshotTransactionCounter, "",
 806                 DataStoreVersions.CURRENT_VERSION);
 807
 808         createSnapshotTransaction.tell(CreateSnapshot.INSTANCE, self());
 809     }
 810
 811     @VisibleForTesting
 812     @Override
 813     protected void applySnapshot(final byte[] snapshotBytes) {
 814         // Since this will be done only on Recovery or when this actor is a Follower
 815         // we can safely commit everything in here. We not need to worry about event notifications
 816         // as they would have already been disabled on the follower
 817
 818         LOG.info("{}: Applying snapshot", persistenceId());
 819         try {
 820             DOMStoreWriteTransaction transaction = store.newWriteOnlyTransaction();
 821
 822             NormalizedNode<?, ?> node = SerializationUtils.deserializeNormalizedNode(snapshotBytes);
 823
 824             // delete everything first
 825             transaction.delete(DATASTORE_ROOT);
 826
 827             // Add everything from the remote node back
 828             transaction.write(DATASTORE_ROOT, node);
 829             syncCommitTransaction(transaction);
 830         } catch (InterruptedException | ExecutionException e) {
 831             LOG.error("{}: An exception occurred when applying snapshot", persistenceId(), e);
 832         } finally {
 833             LOG.info("{}: Done applying snapshot", persistenceId());
 834         }
 835     }
 836
 837     @Override
 838     protected void onStateChanged() {
 839         boolean isLeader = isLeader();
 840         for (ActorSelection dataChangeListener : dataChangeListeners) {
 841             dataChangeListener.tell(new EnableNotification(isLeader), getSelf());
 842         }
 843
 844         if(isLeader) {
 845             for(DelayedListenerRegistration reg: delayedListenerRegistrations) {
 846                 if(!reg.isClosed()) {
 847                     reg.setDelegate(doChangeListenerRegistration(reg.getRegisterChangeListener()));
 848                 }
 849             }
 850
 851             delayedListenerRegistrations.clear();
 852         }
 853
 854         shardMBean.setRaftState(getRaftState().name());
 855         shardMBean.setCurrentTerm(getCurrentTerm());
 856
 857         // If this actor is no longer the leader close all the transaction chains
 858         if(!isLeader){
 859             for(Map.Entry<String, DOMStoreTransactionChain> entry : transactionChains.entrySet()){
 860                 if(LOG.isDebugEnabled()) {
 861                     LOG.debug(
 862                         "{}: onStateChanged: Closing transaction chain {} because shard {} is no longer the leader",
 863                         persistenceId(), entry.getKey(), getId());
 864                 }
 865                 entry.getValue().close();
 866             }
 867
 868             transactionChains.clear();
 869         }
 870     }
 871
 872     @Override
 873     protected DataPersistenceProvider persistence() {
 874         return dataPersistenceProvider;
 875     }
 876
 877     @Override protected void onLeaderChanged(final String oldLeader, final String newLeader) {
 878         shardMBean.setLeader(newLeader);
 879     }
 880
 881     @Override public String persistenceId() {
 882         return this.name.toString();
 883     }
 884
 885     @VisibleForTesting
 886     DataPersistenceProvider getDataPersistenceProvider() {
 887         return dataPersistenceProvider;
 888     }
 889
 890     private static class ShardCreator implements Creator<Shard> {
 891
 892         private static final long serialVersionUID = 1L;
 893
 894         final ShardIdentifier name;
 895         final Map<ShardIdentifier, String> peerAddresses;
 896         final DatastoreContext datastoreContext;
 897         final SchemaContext schemaContext;
 898
 899         ShardCreator(final ShardIdentifier name, final Map<ShardIdentifier, String> peerAddresses,
 900                 final DatastoreContext datastoreContext, final SchemaContext schemaContext) {
 901             this.name = name;
 902             this.peerAddresses = peerAddresses;
 903             this.datastoreContext = datastoreContext;
 904             this.schemaContext = schemaContext;
 905         }
 906
 907         @Override
 908         public Shard create() throws Exception {
 909             return new Shard(name, peerAddresses, datastoreContext, schemaContext);
 910         }
 911     }
 912
 913     @VisibleForTesting
 914     InMemoryDOMDataStore getDataStore() {
 915         return store;
 916     }
 917
 918     @VisibleForTesting
 919     ShardStats getShardMBean() {
 920         return shardMBean;
 921     }
 922
 923     private static class DelayedListenerRegistration implements
 924         ListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>> {
 925
 926         private volatile boolean closed;
 927
 928         private final RegisterChangeListener registerChangeListener;
 929
 930         private volatile ListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier,
 931                                                              NormalizedNode<?, ?>>> delegate;
 932
 933         DelayedListenerRegistration(final RegisterChangeListener registerChangeListener) {
 934             this.registerChangeListener = registerChangeListener;
 935         }
 936
 937         void setDelegate( final ListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier,
 938                                             NormalizedNode<?, ?>>> registration) {
 939             this.delegate = registration;
 940         }
 941
 942         boolean isClosed() {
 943             return closed;
 944         }
 945
 946         RegisterChangeListener getRegisterChangeListener() {
 947             return registerChangeListener;
 948         }
 949
 950         @Override
 951         public AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>> getInstance() {
 952             return delegate != null ? delegate.getInstance() : null;
 953         }
 954
 955         @Override
 956         public void close() {
 957             closed = true;
 958             if(delegate != null) {
 959                 delegate.close();
 960             }
 961         }
 962     }
 963 }