X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FShard.java;h=6da9fe947a0651e5d96981b87a274bde814f902a;hp=ff0f4592cba972261660c5ea1db2c3e415a02608;hb=103ceecd0195cca6c87fbd62a687d8addf128784;hpb=f3473ee42d45f1524dcafa6cc37e19e0393e9693 diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java index ff0f4592cb..6da9fe947a 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java @@ -18,23 +18,17 @@ import akka.serialization.Serialization; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Optional; import com.google.common.base.Preconditions; -import com.google.common.collect.Lists; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import java.io.IOException; -import java.util.Collection; import java.util.HashMap; -import java.util.List; import java.util.Map; -import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import javax.annotation.Nonnull; -import org.opendaylight.controller.cluster.DataPersistenceProvider; import org.opendaylight.controller.cluster.common.actor.CommonConfig; import org.opendaylight.controller.cluster.common.actor.MeteringBehavior; import org.opendaylight.controller.cluster.datastore.ShardCommitCoordinator.CohortEntry; -import org.opendaylight.controller.cluster.datastore.compat.BackwardsCompatibleThreePhaseCommitCohort; import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException; import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier; import org.opendaylight.controller.cluster.datastore.identifiers.ShardTransactionIdentifier; @@ -43,46 +37,39 @@ import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats import org.opendaylight.controller.cluster.datastore.messages.AbortTransaction; import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply; import org.opendaylight.controller.cluster.datastore.messages.ActorInitialized; +import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications; import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction; import org.opendaylight.controller.cluster.datastore.messages.CloseTransactionChain; import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction; import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply; -import org.opendaylight.controller.cluster.datastore.messages.CreateSnapshot; import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction; import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply; -import org.opendaylight.controller.cluster.datastore.messages.EnableNotification; import org.opendaylight.controller.cluster.datastore.messages.ForwardedReadyTransaction; import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved; -import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply; +import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransaction; import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener; -import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply; +import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListener; +import org.opendaylight.controller.cluster.datastore.messages.ShardLeaderStateChanged; import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext; import org.opendaylight.controller.cluster.datastore.modification.Modification; import org.opendaylight.controller.cluster.datastore.modification.ModificationPayload; import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification; import org.opendaylight.controller.cluster.datastore.utils.Dispatchers; import org.opendaylight.controller.cluster.datastore.utils.MessageTracker; -import org.opendaylight.controller.cluster.datastore.utils.SerializationUtils; +import org.opendaylight.controller.cluster.notifications.LeaderStateChanged; import org.opendaylight.controller.cluster.notifications.RegisterRoleChangeListener; import org.opendaylight.controller.cluster.notifications.RoleChangeNotifier; import org.opendaylight.controller.cluster.raft.RaftActor; -import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry; +import org.opendaylight.controller.cluster.raft.RaftActorRecoveryCohort; +import org.opendaylight.controller.cluster.raft.RaftActorSnapshotCohort; import org.opendaylight.controller.cluster.raft.base.messages.FollowerInitialSyncUpStatus; import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply; import org.opendaylight.controller.cluster.raft.protobuff.client.messages.CompositeModificationByteStringPayload; import org.opendaylight.controller.cluster.raft.protobuff.client.messages.CompositeModificationPayload; -import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload; -import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener; -import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore; -import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStoreFactory; -import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort; -import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransaction; -import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain; -import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionFactory; -import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction; -import org.opendaylight.yangtools.concepts.ListenerRegistration; -import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; -import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; +import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree; +import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate; +import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailedException; +import org.opendaylight.yangtools.yang.data.api.schema.tree.ModificationType; import org.opendaylight.yangtools.yang.model.api.SchemaContext; import scala.concurrent.duration.Duration; import scala.concurrent.duration.FiniteDuration; @@ -90,39 +77,29 @@ import scala.concurrent.duration.FiniteDuration; /** * A Shard represents a portion of the logical data tree
*

- * Our Shard uses InMemoryDataStore as it's internal representation and delegates all requests it + * Our Shard uses InMemoryDataTree as it's internal representation and delegates all requests it *

*/ public class Shard extends RaftActor { - private static final YangInstanceIdentifier DATASTORE_ROOT = YangInstanceIdentifier.builder().build(); - private static final Object TX_COMMIT_TIMEOUT_CHECK_MESSAGE = "txCommitTimeoutCheck"; + @VisibleForTesting + static final Object GET_SHARD_MBEAN_MESSAGE = "getShardMBeanMessage"; + @VisibleForTesting static final String DEFAULT_NAME = "default"; // The state of this Shard - private final InMemoryDOMDataStore store; + private final ShardDataTree store; /// The name of this shard - private final ShardIdentifier name; + private final String name; private final ShardStats shardMBean; - private final List dataChangeListeners = Lists.newArrayList(); - - private final List delayedListenerRegistrations = - Lists.newArrayList(); - private DatastoreContext datastoreContext; - private DataPersistenceProvider dataPersistenceProvider; - - private SchemaContext schemaContext; - - private int createSnapshotTransactionCounter; - private final ShardCommitCoordinator commitCoordinator; private long transactionCommitTimeout; @@ -133,53 +110,38 @@ public class Shard extends RaftActor { private final MessageTracker appendEntriesReplyTracker; - private final ReadyTransactionReply READY_TRANSACTION_REPLY = new ReadyTransactionReply( - Serialization.serializedActorPath(getSelf())); + private final ShardTransactionActorFactory transactionActorFactory; + private final ShardSnapshotCohort snapshotCohort; - /** - * Coordinates persistence recovery on startup. - */ - private ShardRecoveryCoordinator recoveryCoordinator; - private List currentLogRecoveryBatch; + private final DataTreeChangeListenerSupport treeChangeSupport = new DataTreeChangeListenerSupport(this); + private final DataChangeListenerSupport changeSupport = new DataChangeListenerSupport(this); - private final Map transactionChains = new HashMap<>(); - - private final String txnDispatcherPath; - - protected Shard(final ShardIdentifier name, final Map peerAddresses, + protected Shard(final ShardIdentifier name, final Map peerAddresses, final DatastoreContext datastoreContext, final SchemaContext schemaContext) { - super(name.toString(), mapPeerAddresses(peerAddresses), - Optional.of(datastoreContext.getShardRaftConfig())); + super(name.toString(), new HashMap<>(peerAddresses), Optional.of(datastoreContext.getShardRaftConfig()), + DataStoreVersions.CURRENT_VERSION); - this.name = name; + this.name = name.toString(); this.datastoreContext = datastoreContext; - this.schemaContext = schemaContext; - this.dataPersistenceProvider = (datastoreContext.isPersistent()) - ? new PersistentDataProvider() : new NonPersistentRaftDataProvider(); - this.txnDispatcherPath = new Dispatchers(context().system().dispatchers()) - .getDispatcherPath(Dispatchers.DispatcherType.Transaction); + setPersistence(datastoreContext.isPersistent()); LOG.info("Shard created : {}, persistent : {}", name, datastoreContext.isPersistent()); - store = InMemoryDOMDataStoreFactory.create(name.toString(), null, - datastoreContext.getDataStoreProperties()); - - if(schemaContext != null) { - store.onGlobalContextUpdated(schemaContext); - } + store = new ShardDataTree(schemaContext); shardMBean = ShardMBeanFactory.getShardStatsMBean(name.toString(), datastoreContext.getDataStoreMXBeanType()); - shardMBean.setNotificationManager(store.getDataChangeListenerNotificationManager()); + shardMBean.setShardActor(getSelf()); if (isMetricsCaptureEnabled()) { getContext().become(new MeteringBehavior(this)); } - commitCoordinator = new ShardCommitCoordinator(TimeUnit.SECONDS.convert(1, TimeUnit.MINUTES), - datastoreContext.getShardTransactionCommitQueueCapacity(), LOG, name.toString()); + commitCoordinator = new ShardCommitCoordinator(store, + datastoreContext.getShardCommitQueueExpiryTimeoutInMillis(), + datastoreContext.getShardTransactionCommitQueueCapacity(), self(), LOG, this.name); setTransactionCommitTimeout(); @@ -188,27 +150,21 @@ public class Shard extends RaftActor { appendEntriesReplyTracker = new MessageTracker(AppendEntriesReply.class, getRaftActorContext().getConfigParams().getIsolatedCheckIntervalInMillis()); + + transactionActorFactory = new ShardTransactionActorFactory(store, datastoreContext, + new Dispatchers(context().system().dispatchers()).getDispatcherPath( + Dispatchers.DispatcherType.Transaction), self(), getContext(), shardMBean); + + snapshotCohort = new ShardSnapshotCohort(transactionActorFactory, store, LOG, this.name); } private void setTransactionCommitTimeout() { transactionCommitTimeout = TimeUnit.MILLISECONDS.convert( - datastoreContext.getShardTransactionCommitTimeoutInSeconds(), TimeUnit.SECONDS); - } - - private static Map mapPeerAddresses( - final Map peerAddresses) { - Map map = new HashMap<>(); - - for (Map.Entry entry : peerAddresses - .entrySet()) { - map.put(entry.getKey().toString(), entry.getValue()); - } - - return map; + datastoreContext.getShardTransactionCommitTimeoutInSeconds(), TimeUnit.SECONDS) / 2; } public static Props props(final ShardIdentifier name, - final Map peerAddresses, + final Map peerAddresses, final DatastoreContext datastoreContext, final SchemaContext schemaContext) { Preconditions.checkNotNull(name, "name should not be null"); Preconditions.checkNotNull(peerAddresses, "peerAddresses should not be null"); @@ -272,8 +228,13 @@ public class Shard extends RaftActor { try { if (CreateTransaction.SERIALIZABLE_CLASS.isInstance(message)) { handleCreateTransaction(message); + } else if (BatchedModifications.class.isInstance(message)) { + handleBatchedModifications((BatchedModifications)message); } else if (message instanceof ForwardedReadyTransaction) { - handleForwardedReadyTransaction((ForwardedReadyTransaction) message); + commitCoordinator.handleForwardedReadyTransaction((ForwardedReadyTransaction) message, + getSender(), this); + } else if (message instanceof ReadyLocalTransaction) { + handleReadyLocalTransaction((ReadyLocalTransaction)message); } else if (CanCommitTransaction.SERIALIZABLE_CLASS.isInstance(message)) { handleCanCommitTransaction(CanCommitTransaction.fromSerializable(message)); } else if (CommitTransaction.SERIALIZABLE_CLASS.isInstance(message)) { @@ -283,7 +244,9 @@ public class Shard extends RaftActor { } else if (CloseTransactionChain.SERIALIZABLE_CLASS.isInstance(message)) { closeTransactionChain(CloseTransactionChain.fromSerializable(message)); } else if (message instanceof RegisterChangeListener) { - registerChangeListener((RegisterChangeListener) message); + changeSupport.onMessage((RegisterChangeListener) message, isLeader()); + } else if (message instanceof RegisterDataTreeChangeListener) { + treeChangeSupport.onMessage((RegisterDataTreeChangeListener) message, isLeader()); } else if (message instanceof UpdateSchemaContext) { updateSchemaContext((UpdateSchemaContext) message); } else if (message instanceof PeerAddressResolved) { @@ -296,9 +259,11 @@ public class Shard extends RaftActor { onDatastoreContext((DatastoreContext)message); } else if(message instanceof RegisterRoleChangeListener){ roleChangeNotifier.get().forward(message, context()); - } else if (message instanceof FollowerInitialSyncUpStatus){ + } else if (message instanceof FollowerInitialSyncUpStatus) { shardMBean.setFollowerInitialSyncStatus(((FollowerInitialSyncUpStatus) message).isInitialSyncDone()); context().parent().tell(message, self()); + } else if(GET_SHARD_MBEAN_MESSAGE.equals(message)){ + sender().tell(getShardMBean(), self()); } else { super.onReceiveCommand(message); } @@ -312,6 +277,12 @@ public class Shard extends RaftActor { return roleChangeNotifier; } + @Override + protected LeaderStateChanged newLeaderStateChanged(String memberId, String leaderId) { + return new ShardLeaderStateChanged(memberId, leaderId, + isLeader() ? Optional.of(store.getDataTree()) : Optional.absent()); + } + private void onDatastoreContext(DatastoreContext context) { datastoreContext = context; @@ -319,12 +290,10 @@ public class Shard extends RaftActor { setTransactionCommitTimeout(); - if(datastoreContext.isPersistent() && - dataPersistenceProvider instanceof NonPersistentRaftDataProvider) { - dataPersistenceProvider = new PersistentDataProvider(); - } else if(!datastoreContext.isPersistent() && - dataPersistenceProvider instanceof PersistentDataProvider) { - dataPersistenceProvider = new NonPersistentRaftDataProvider(); + if(datastoreContext.isPersistent() && !persistence().isRecoveryApplicable()) { + setPersistence(true); + } else if(!datastoreContext.isPersistent() && persistence().isRecoveryApplicable()) { + setPersistence(false); } updateConfigParams(datastoreContext.getShardRaftConfig()); @@ -333,64 +302,64 @@ public class Shard extends RaftActor { private void handleTransactionCommitTimeoutCheck() { CohortEntry cohortEntry = commitCoordinator.getCurrentCohortEntry(); if(cohortEntry != null) { - long elapsed = System.currentTimeMillis() - cohortEntry.getLastAccessTime(); - if(elapsed > transactionCommitTimeout) { + if(cohortEntry.isExpired(transactionCommitTimeout)) { LOG.warn("{}: Current transaction {} has timed out after {} ms - aborting", persistenceId(), cohortEntry.getTransactionID(), transactionCommitTimeout); doAbortTransaction(cohortEntry.getTransactionID(), null); } } + + commitCoordinator.cleanupExpiredCohortEntries(); } - private void handleCommitTransaction(final CommitTransaction commit) { - final String transactionID = commit.getTransactionID(); + private static boolean isEmptyCommit(final DataTreeCandidate candidate) { + return ModificationType.UNMODIFIED.equals(candidate.getRootNode().getModificationType()); + } - LOG.debug("{}: Committing transaction {}", persistenceId(), transactionID); + void continueCommit(final CohortEntry cohortEntry) throws Exception { + final DataTreeCandidate candidate = cohortEntry.getCohort().getCandidate(); - // Get the current in-progress cohort entry in the commitCoordinator if it corresponds to - // this transaction. - final CohortEntry cohortEntry = commitCoordinator.getCohortEntryIfCurrent(transactionID); - if(cohortEntry == null) { - // We're not the current Tx - the Tx was likely expired b/c it took too long in - // between the canCommit and commit messages. - IllegalStateException ex = new IllegalStateException( - String.format("%s: Cannot commit transaction %s - it is not the current transaction", - persistenceId(), transactionID)); - LOG.error(ex.getMessage()); + // If we do not have any followers and we are not using persistence + // or if cohortEntry has no modifications + // we can apply modification to the state immediately + if ((!hasFollowers() && !persistence().isRecoveryApplicable()) || isEmptyCommit(candidate)) { + applyModificationToState(cohortEntry.getReplySender(), cohortEntry.getTransactionID(), candidate); + } else { + Shard.this.persistData(cohortEntry.getReplySender(), cohortEntry.getTransactionID(), + DataTreeCandidatePayload.create(candidate)); + } + } + + private void handleCommitTransaction(final CommitTransaction commit) { + if(!commitCoordinator.handleCommit(commit.getTransactionID(), getSender(), this)) { shardMBean.incrementFailedTransactionsCount(); - getSender().tell(new akka.actor.Status.Failure(ex), getSelf()); - return; } + } - // We perform the preCommit phase here atomically with the commit phase. This is an - // optimization to eliminate the overhead of an extra preCommit message. We lose front-end - // coordination of preCommit across shards in case of failure but preCommit should not - // normally fail since we ensure only one concurrent 3-phase commit. + private void finishCommit(@Nonnull final ActorRef sender, @Nonnull final String transactionID, @Nonnull final CohortEntry cohortEntry) { + LOG.debug("{}: Finishing commit for transaction {}", persistenceId(), cohortEntry.getTransactionID()); try { // We block on the future here so we don't have to worry about possibly accessing our // state on a different thread outside of our dispatcher. Also, the data store // currently uses a same thread executor anyway. - cohortEntry.getCohort().preCommit().get(); + cohortEntry.getCohort().commit().get(); + + sender.tell(CommitTransactionReply.INSTANCE.toSerializable(), getSelf()); + + shardMBean.incrementCommittedTransactionCount(); + shardMBean.setLastCommittedTransactionTime(System.currentTimeMillis()); - // If we do not have any followers and we are not using persistence - // or if cohortEntry has no modifications - // we can apply modification to the state immediately - if((!hasFollowers() && !persistence().isRecoveryApplicable()) || (!cohortEntry.hasModifications())){ - applyModificationToState(getSender(), transactionID, cohortEntry.getModification()); - } else { - Shard.this.persistData(getSender(), transactionID, - new ModificationPayload(cohortEntry.getModification())); - } } catch (Exception e) { - LOG.error("{} An exception occurred while preCommitting transaction {}", - persistenceId(), cohortEntry.getTransactionID(), e); + sender.tell(new akka.actor.Status.Failure(e), getSelf()); + + LOG.error("{}, An exception occurred while committing transaction {}", persistenceId(), + transactionID, e); shardMBean.incrementFailedTransactionsCount(); - getSender().tell(new akka.actor.Status.Failure(e), getSelf()); + } finally { + commitCoordinator.currentTransactionComplete(transactionID, true); } - - cohortEntry.updateLastAccessTime(); } private void finishCommit(@Nonnull final ActorRef sender, final @Nonnull String transactionID) { @@ -398,7 +367,7 @@ public class Shard extends RaftActor { // after the commit has been replicated to a majority of the followers. CohortEntry cohortEntry = commitCoordinator.getCohortEntryIfCurrent(transactionID); - if(cohortEntry == null) { + if (cohortEntry == null) { // The transaction is no longer the current commit. This can happen if the transaction // was aborted prior, most likely due to timeout in the front-end. We need to finish // committing the transaction though since it was successfully persisted and replicated @@ -407,7 +376,13 @@ public class Shard extends RaftActor { // transaction. cohortEntry = commitCoordinator.getAndRemoveCohortEntry(transactionID); if(cohortEntry != null) { - commitWithNewTransaction(cohortEntry.getModification()); + try { + store.applyForeignCandidate(transactionID, cohortEntry.getCohort().getCandidate()); + } catch (DataValidationFailedException e) { + shardMBean.incrementFailedTransactionsCount(); + LOG.error("{}: Failed to re-apply transaction {}", persistenceId(), transactionID, e); + } + sender.tell(CommitTransactionReply.INSTANCE.toSerializable(), getSelf()); } else { // This really shouldn't happen - it likely means that persistence or replication @@ -418,68 +393,77 @@ public class Shard extends RaftActor { LOG.error(ex.getMessage()); sender.tell(new akka.actor.Status.Failure(ex), getSelf()); } - - return; - } - - LOG.debug("{}: Finishing commit for transaction {}", persistenceId(), cohortEntry.getTransactionID()); - - try { - // We block on the future here so we don't have to worry about possibly accessing our - // state on a different thread outside of our dispatcher. Also, the data store - // currently uses a same thread executor anyway. - cohortEntry.getCohort().commit().get(); - - sender.tell(CommitTransactionReply.INSTANCE.toSerializable(), getSelf()); - - shardMBean.incrementCommittedTransactionCount(); - shardMBean.setLastCommittedTransactionTime(System.currentTimeMillis()); - - } catch (Exception e) { - sender.tell(new akka.actor.Status.Failure(e), getSelf()); - - LOG.error("{}, An exception occurred while committing transaction {}", persistenceId(), - transactionID, e); - shardMBean.incrementFailedTransactionsCount(); - } finally { - commitCoordinator.currentTransactionComplete(transactionID, true); + } else { + finishCommit(sender, transactionID, cohortEntry); } } private void handleCanCommitTransaction(final CanCommitTransaction canCommit) { LOG.debug("{}: Can committing transaction {}", persistenceId(), canCommit.getTransactionID()); - commitCoordinator.handleCanCommit(canCommit, getSender(), self()); + commitCoordinator.handleCanCommit(canCommit.getTransactionID(), getSender(), this); + } + + private void noLeaderError(Object message) { + // TODO: rather than throwing an immediate exception, we could schedule a timer to try again to make + // it more resilient in case we're in the process of electing a new leader. + getSender().tell(new akka.actor.Status.Failure(new NoShardLeaderException(String.format( + "Could not find the leader for shard %s. This typically happens" + + " when the system is coming up or recovering and a leader is being elected. Try again" + + " later.", persistenceId()))), getSelf()); + } + + private void handleBatchedModifications(BatchedModifications batched) { + // This message is sent to prepare the modificationsa transaction directly on the Shard as an + // optimization to avoid the extra overhead of a separate ShardTransaction actor. On the last + // BatchedModifications message, the caller sets the ready flag in the message indicating + // modifications are complete. The reply contains the cohort actor path (this actor) for the caller + // to initiate the 3-phase commit. This also avoids the overhead of sending an additional + // ReadyTransaction message. + + // If we're not the leader then forward to the leader. This is a safety measure - we shouldn't + // normally get here if we're not the leader as the front-end (TransactionProxy) should determine + // the primary/leader shard. However with timing and caching on the front-end, there's a small + // window where it could have a stale leader during leadership transitions. + // + if(isLeader()) { + try { + commitCoordinator.handleBatchedModifications(batched, getSender(), this); + } catch (Exception e) { + LOG.error("{}: Error handling BatchedModifications for Tx {}", persistenceId(), + batched.getTransactionID(), e); + getSender().tell(new akka.actor.Status.Failure(e), getSelf()); + } + } else { + ActorSelection leader = getLeader(); + if(leader != null) { + // TODO: what if this is not the first batch and leadership changed in between batched messages? + // We could check if the commitCoordinator already has a cached entry and forward all the previous + // batched modifications. + LOG.debug("{}: Forwarding BatchedModifications to leader {}", persistenceId(), leader); + leader.forward(batched, getContext()); + } else { + noLeaderError(batched); + } + } } - private void handleForwardedReadyTransaction(ForwardedReadyTransaction ready) { - LOG.debug("{}: Readying transaction {}, client version {}", persistenceId(), - ready.getTransactionID(), ready.getTxnClientVersion()); - - // This message is forwarded by the ShardTransaction on ready. We cache the cohort in the - // commitCoordinator in preparation for the subsequent three phase commit initiated by - // the front-end. - commitCoordinator.transactionReady(ready.getTransactionID(), ready.getCohort(), - ready.getModification()); - - // Return our actor path as we'll handle the three phase commit, except if the Tx client - // version < 1 (Helium-1 version). This means the Tx was initiated by a base Helium version - // node. In that case, the subsequent 3-phase commit messages won't contain the - // transactionId so to maintain backwards compatibility, we create a separate cohort actor - // to provide the compatible behavior. - if(ready.getTxnClientVersion() < DataStoreVersions.HELIUM_1_VERSION) { - LOG.debug("{}: Creating BackwardsCompatibleThreePhaseCommitCohort", persistenceId()); - ActorRef replyActorPath = getContext().actorOf(BackwardsCompatibleThreePhaseCommitCohort.props( - ready.getTransactionID())); - - ReadyTransactionReply readyTransactionReply = - new ReadyTransactionReply(Serialization.serializedActorPath(replyActorPath)); - getSender().tell(ready.isReturnSerialized() ? readyTransactionReply.toSerializable() : - readyTransactionReply, getSelf()); - + private void handleReadyLocalTransaction(final ReadyLocalTransaction message) { + if (isLeader()) { + try { + commitCoordinator.handleReadyLocalTransaction(message, getSender(), this); + } catch (Exception e) { + LOG.error("{}: Error handling LocalModifications for Tx {}", persistenceId(), + message.getTransactionID(), e); + getSender().tell(new akka.actor.Status.Failure(e), getSelf()); + } } else { - - getSender().tell(ready.isReturnSerialized() ? READY_TRANSACTION_REPLY.toSerializable() : - READY_TRANSACTION_REPLY, getSelf()); + ActorSelection leader = getLeader(); + if (leader != null) { + LOG.debug("{}: Forwarding LocalModifications to leader {}", persistenceId(), leader); + leader.forward(message, getContext()); + } else { + noLeaderError(message); + } } } @@ -536,67 +520,15 @@ public class Shard extends RaftActor { } private void closeTransactionChain(final CloseTransactionChain closeTransactionChain) { - DOMStoreTransactionChain chain = - transactionChains.remove(closeTransactionChain.getTransactionChainId()); - - if(chain != null) { - chain.close(); - } + store.closeTransactionChain(closeTransactionChain.getTransactionChainId()); } private ActorRef createTypedTransactionActor(int transactionType, ShardTransactionIdentifier transactionId, String transactionChainId, short clientVersion ) { - DOMStoreTransactionFactory factory = store; - - if(!transactionChainId.isEmpty()) { - factory = transactionChains.get(transactionChainId); - if(factory == null){ - DOMStoreTransactionChain transactionChain = store.createTransactionChain(); - transactionChains.put(transactionChainId, transactionChain); - factory = transactionChain; - } - } - - if(this.schemaContext == null) { - throw new IllegalStateException("SchemaContext is not set"); - } - - if (transactionType == TransactionProxy.TransactionType.WRITE_ONLY.ordinal()) { - - shardMBean.incrementWriteOnlyTransactionCount(); - - return createShardTransaction(factory.newWriteOnlyTransaction(), transactionId, clientVersion); - - } else if (transactionType == TransactionProxy.TransactionType.READ_WRITE.ordinal()) { - - shardMBean.incrementReadWriteTransactionCount(); - - return createShardTransaction(factory.newReadWriteTransaction(), transactionId, clientVersion); - - } else if (transactionType == TransactionProxy.TransactionType.READ_ONLY.ordinal()) { - - shardMBean.incrementReadOnlyTransactionCount(); - - return createShardTransaction(factory.newReadOnlyTransaction(), transactionId, clientVersion); - - } else { - throw new IllegalArgumentException( - "Shard="+name + ":CreateTransaction message has unidentified transaction type=" - + transactionType); - } - } - - private ActorRef createShardTransaction(DOMStoreTransaction transaction, ShardTransactionIdentifier transactionId, - short clientVersion){ - return getContext().actorOf( - ShardTransaction.props(transaction, getSelf(), - schemaContext, datastoreContext, shardMBean, - transactionId.getRemoteTransactionId(), clientVersion) - .withDispatcher(txnDispatcherPath), - transactionId.toString()); - + return transactionActorFactory.newShardTransaction(TransactionType.fromInt(transactionType), + transactionId, transactionChainId, clientVersion); } private void createTransaction(CreateTransaction createTransaction) { @@ -628,173 +560,46 @@ public class Shard extends RaftActor { return transactionActor; } - private void syncCommitTransaction(final DOMStoreWriteTransaction transaction) - throws ExecutionException, InterruptedException { - DOMStoreThreePhaseCommitCohort commitCohort = transaction.ready(); - commitCohort.preCommit().get(); - commitCohort.commit().get(); - } - private void commitWithNewTransaction(final Modification modification) { - DOMStoreWriteTransaction tx = store.newWriteOnlyTransaction(); - modification.apply(tx); + ReadWriteShardDataTreeTransaction tx = store.newReadWriteTransaction(modification.toString(), null); + modification.apply(tx.getSnapshot()); try { - syncCommitTransaction(tx); + snapshotCohort.syncCommitTransaction(tx); shardMBean.incrementCommittedTransactionCount(); shardMBean.setLastCommittedTransactionTime(System.currentTimeMillis()); - } catch (InterruptedException | ExecutionException e) { + } catch (Exception e) { shardMBean.incrementFailedTransactionsCount(); LOG.error("{}: Failed to commit", persistenceId(), e); } } private void updateSchemaContext(final UpdateSchemaContext message) { - this.schemaContext = message.getSchemaContext(); updateSchemaContext(message.getSchemaContext()); - store.onGlobalContextUpdated(message.getSchemaContext()); } @VisibleForTesting void updateSchemaContext(final SchemaContext schemaContext) { - store.onGlobalContextUpdated(schemaContext); + store.updateSchemaContext(schemaContext); } - private void registerChangeListener(final RegisterChangeListener registerChangeListener) { - - LOG.debug("{}: registerDataChangeListener for {}", persistenceId(), registerChangeListener.getPath()); - - ListenerRegistration>> registration; - if(isLeader()) { - registration = doChangeListenerRegistration(registerChangeListener); - } else { - LOG.debug("{}: Shard is not the leader - delaying registration", persistenceId()); - - DelayedListenerRegistration delayedReg = - new DelayedListenerRegistration(registerChangeListener); - delayedListenerRegistrations.add(delayedReg); - registration = delayedReg; - } - - ActorRef listenerRegistration = getContext().actorOf( - DataChangeListenerRegistration.props(registration)); - - LOG.debug("{}: registerDataChangeListener sending reply, listenerRegistrationPath = {} ", - persistenceId(), listenerRegistration.path()); - - getSender().tell(new RegisterChangeListenerReply(listenerRegistration.path()), getSelf()); - } - - private ListenerRegistration>> doChangeListenerRegistration( - final RegisterChangeListener registerChangeListener) { - - ActorSelection dataChangeListenerPath = getContext().system().actorSelection( - registerChangeListener.getDataChangeListenerPath()); - - // Notify the listener if notifications should be enabled or not - // If this shard is the leader then it will enable notifications else - // it will not - dataChangeListenerPath.tell(new EnableNotification(true), getSelf()); - - // Now store a reference to the data change listener so it can be notified - // at a later point if notifications should be enabled or disabled - dataChangeListeners.add(dataChangeListenerPath); - - AsyncDataChangeListener> listener = - new DataChangeListenerProxy(dataChangeListenerPath); - - LOG.debug("{}: Registering for path {}", persistenceId(), registerChangeListener.getPath()); - - return store.registerChangeListener(registerChangeListener.getPath(), listener, - registerChangeListener.getScope()); - } - - private boolean isMetricsCaptureEnabled(){ + private boolean isMetricsCaptureEnabled() { CommonConfig config = new CommonConfig(getContext().system().settings().config()); return config.isMetricCaptureEnabled(); } @Override - protected - void startLogRecoveryBatch(final int maxBatchSize) { - currentLogRecoveryBatch = Lists.newArrayListWithCapacity(maxBatchSize); - - if(LOG.isDebugEnabled()) { - LOG.debug("{}: starting log recovery batch with max size {}", persistenceId(), maxBatchSize); - } + protected RaftActorSnapshotCohort getRaftActorSnapshotCohort() { + return snapshotCohort; } @Override - protected void appendRecoveredLogEntry(final Payload data) { - if(data instanceof ModificationPayload) { - try { - currentLogRecoveryBatch.add(((ModificationPayload) data).getModification()); - } catch (ClassNotFoundException | IOException e) { - LOG.error("{}: Error extracting ModificationPayload", persistenceId(), e); - } - } else if (data instanceof CompositeModificationPayload) { - currentLogRecoveryBatch.add(((CompositeModificationPayload) data).getModification()); - } else if (data instanceof CompositeModificationByteStringPayload) { - currentLogRecoveryBatch.add(((CompositeModificationByteStringPayload) data).getModification()); - } else { - LOG.error("{}: Unknown state received {} during recovery", persistenceId(), data); - } - } - - @Override - protected void applyRecoverySnapshot(final byte[] snapshotBytes) { - if(recoveryCoordinator == null) { - recoveryCoordinator = new ShardRecoveryCoordinator(persistenceId(), schemaContext, - LOG, name.toString()); - } - - recoveryCoordinator.submit(snapshotBytes, store.newWriteOnlyTransaction()); - - if(LOG.isDebugEnabled()) { - LOG.debug("{}: submitted recovery sbapshot", persistenceId()); - } - } - - @Override - protected void applyCurrentLogRecoveryBatch() { - if(recoveryCoordinator == null) { - recoveryCoordinator = new ShardRecoveryCoordinator(persistenceId(), schemaContext, - LOG, name.toString()); - } - - recoveryCoordinator.submit(currentLogRecoveryBatch, store.newWriteOnlyTransaction()); - - if(LOG.isDebugEnabled()) { - LOG.debug("{}: submitted log recovery batch with size {}", persistenceId(), - currentLogRecoveryBatch.size()); - } + @Nonnull + protected RaftActorRecoveryCohort getRaftActorRecoveryCohort() { + return new ShardRecoveryCoordinator(store, persistenceId(), LOG); } @Override protected void onRecoveryComplete() { - if(recoveryCoordinator != null) { - Collection txList = recoveryCoordinator.getTransactions(); - - if(LOG.isDebugEnabled()) { - LOG.debug("{}: recovery complete - committing {} Tx's", persistenceId(), txList.size()); - } - - for(DOMStoreWriteTransaction tx: txList) { - try { - syncCommitTransaction(tx); - shardMBean.incrementCommittedTransactionCount(); - } catch (InterruptedException | ExecutionException e) { - shardMBean.incrementFailedTransactionsCount(); - LOG.error("{}: Failed to commit", persistenceId(), e); - } - } - } - - recoveryCoordinator = null; - currentLogRecoveryBatch = null; - updateJournalStats(); - //notify shard manager getContext().parent().tell(new ActorInitialized(), getSelf()); @@ -811,15 +616,25 @@ public class Shard extends RaftActor { @Override protected void applyState(final ActorRef clientActor, final String identifier, final Object data) { - - if(data instanceof ModificationPayload) { + if (data instanceof DataTreeCandidatePayload) { + if (clientActor == null) { + // No clientActor indicates a replica coming from the leader + try { + store.applyForeignCandidate(identifier, ((DataTreeCandidatePayload)data).getCandidate()); + } catch (DataValidationFailedException | IOException e) { + LOG.error("{}: Error applying replica {}", persistenceId(), identifier, e); + } + } else { + // Replication consensus reached, proceed to commit + finishCommit(clientActor, identifier); + } + } else if (data instanceof ModificationPayload) { try { applyModificationToState(clientActor, identifier, ((ModificationPayload) data).getModification()); } catch (ClassNotFoundException | IOException e) { LOG.error("{}: Error extracting ModificationPayload", persistenceId(), e); } - } - else if (data instanceof CompositeModificationPayload) { + } else if (data instanceof CompositeModificationPayload) { Object modification = ((CompositeModificationPayload) data).getModification(); applyModificationToState(clientActor, identifier, modification); @@ -832,9 +647,6 @@ public class Shard extends RaftActor { persistenceId(), data, data.getClass().getClassLoader(), CompositeModificationPayload.class.getClassLoader()); } - - updateJournalStats(); - } private void applyModificationToState(ActorRef clientActor, String identifier, Object modification) { @@ -852,122 +664,50 @@ public class Shard extends RaftActor { } } - private void updateJournalStats() { - ReplicatedLogEntry lastLogEntry = getLastLogEntry(); - - if (lastLogEntry != null) { - shardMBean.setLastLogIndex(lastLogEntry.getIndex()); - shardMBean.setLastLogTerm(lastLogEntry.getTerm()); - } - - shardMBean.setCommitIndex(getCommitIndex()); - shardMBean.setLastApplied(getLastApplied()); - shardMBean.setInMemoryJournalDataSize(getRaftActorContext().getReplicatedLog().dataSize()); - } - - @Override - protected void createSnapshot() { - // Create a transaction actor. We are really going to treat the transaction as a worker - // so that this actor does not get block building the snapshot. THe transaction actor will - // after processing the CreateSnapshot message. - - ActorRef createSnapshotTransaction = createTransaction( - TransactionProxy.TransactionType.READ_ONLY.ordinal(), - "createSnapshot" + ++createSnapshotTransactionCounter, "", - DataStoreVersions.CURRENT_VERSION); - - createSnapshotTransaction.tell(CreateSnapshot.INSTANCE, self()); - } - - @VisibleForTesting - @Override - protected void applySnapshot(final byte[] snapshotBytes) { - // Since this will be done only on Recovery or when this actor is a Follower - // we can safely commit everything in here. We not need to worry about event notifications - // as they would have already been disabled on the follower - - LOG.info("{}: Applying snapshot", persistenceId()); - try { - DOMStoreWriteTransaction transaction = store.newWriteOnlyTransaction(); - - NormalizedNode node = SerializationUtils.deserializeNormalizedNode(snapshotBytes); - - // delete everything first - transaction.delete(DATASTORE_ROOT); - - // Add everything from the remote node back - transaction.write(DATASTORE_ROOT, node); - syncCommitTransaction(transaction); - } catch (InterruptedException | ExecutionException e) { - LOG.error("{}: An exception occurred when applying snapshot", persistenceId(), e); - } finally { - LOG.info("{}: Done applying snapshot", persistenceId()); - } - } - @Override protected void onStateChanged() { boolean isLeader = isLeader(); - for (ActorSelection dataChangeListener : dataChangeListeners) { - dataChangeListener.tell(new EnableNotification(isLeader), getSelf()); - } - - if(isLeader) { - for(DelayedListenerRegistration reg: delayedListenerRegistrations) { - if(!reg.isClosed()) { - reg.setDelegate(doChangeListenerRegistration(reg.getRegisterChangeListener())); - } - } - - delayedListenerRegistrations.clear(); - } - - shardMBean.setRaftState(getRaftState().name()); - shardMBean.setCurrentTerm(getCurrentTerm()); + changeSupport.onLeadershipChange(isLeader); + treeChangeSupport.onLeadershipChange(isLeader); // If this actor is no longer the leader close all the transaction chains - if(!isLeader){ - for(Map.Entry entry : transactionChains.entrySet()){ - if(LOG.isDebugEnabled()) { - LOG.debug( - "{}: onStateChanged: Closing transaction chain {} because shard {} is no longer the leader", - persistenceId(), entry.getKey(), getId()); - } - entry.getValue().close(); + if (!isLeader) { + if(LOG.isDebugEnabled()) { + LOG.debug( + "{}: onStateChanged: Closing all transaction chains because shard {} is no longer the leader", + persistenceId(), getId()); } - transactionChains.clear(); + store.closeAllTransactionChains(); } } @Override - protected DataPersistenceProvider persistence() { - return dataPersistenceProvider; - } - - @Override protected void onLeaderChanged(final String oldLeader, final String newLeader) { - shardMBean.setLeader(newLeader); + protected void onLeaderChanged(String oldLeader, String newLeader) { + shardMBean.incrementLeadershipChangeCount(); } - @Override public String persistenceId() { - return this.name.toString(); + @Override + public String persistenceId() { + return this.name; } @VisibleForTesting - DataPersistenceProvider getDataPersistenceProvider() { - return dataPersistenceProvider; + ShardCommitCoordinator getCommitCoordinator() { + return commitCoordinator; } + private static class ShardCreator implements Creator { private static final long serialVersionUID = 1L; final ShardIdentifier name; - final Map peerAddresses; + final Map peerAddresses; final DatastoreContext datastoreContext; final SchemaContext schemaContext; - ShardCreator(final ShardIdentifier name, final Map peerAddresses, + ShardCreator(final ShardIdentifier name, final Map peerAddresses, final DatastoreContext datastoreContext, final SchemaContext schemaContext) { this.name = name; this.peerAddresses = peerAddresses; @@ -982,7 +722,7 @@ public class Shard extends RaftActor { } @VisibleForTesting - public InMemoryDOMDataStore getDataStore() { + public ShardDataTree getDataStore() { return store; } @@ -990,45 +730,4 @@ public class Shard extends RaftActor { ShardStats getShardMBean() { return shardMBean; } - - private static class DelayedListenerRegistration implements - ListenerRegistration>> { - - private volatile boolean closed; - - private final RegisterChangeListener registerChangeListener; - - private volatile ListenerRegistration>> delegate; - - DelayedListenerRegistration(final RegisterChangeListener registerChangeListener) { - this.registerChangeListener = registerChangeListener; - } - - void setDelegate( final ListenerRegistration>> registration) { - this.delegate = registration; - } - - boolean isClosed() { - return closed; - } - - RegisterChangeListener getRegisterChangeListener() { - return registerChangeListener; - } - - @Override - public AsyncDataChangeListener> getInstance() { - return delegate != null ? delegate.getInstance() : null; - } - - @Override - public void close() { - closed = true; - if(delegate != null) { - delegate.close(); - } - } - } }