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=a30b6f7516981411e589577f3fda9a9a0f9bc887;hp=15b97b0f8f1e05c638d2d9f9ad74c34ac2a10eba;hb=4349b034606957d3e876b82b14a292e6739a986a;hpb=fc89f3b66b3e5db73e09792bbecebf3156d70d20 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 15b97b0f8f..a30b6f7516 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 @@ -43,6 +43,8 @@ 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.BatchedModificationsReply; import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction; import org.opendaylight.controller.cluster.datastore.messages.CloseTransactionChain; import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction; @@ -66,7 +68,6 @@ import org.opendaylight.controller.cluster.datastore.utils.SerializationUtils; 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.base.messages.FollowerInitialSyncUpStatus; import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply; import org.opendaylight.controller.cluster.raft.protobuff.client.messages.CompositeModificationByteStringPayload; @@ -77,8 +78,6 @@ 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; @@ -106,7 +105,7 @@ public class Shard extends RaftActor { private final InMemoryDOMDataStore store; /// The name of this shard - private final ShardIdentifier name; + private final String name; private final ShardStats shardMBean; @@ -143,16 +142,15 @@ public class Shard extends RaftActor { private ShardRecoveryCoordinator recoveryCoordinator; private List currentLogRecoveryBatch; - private final Map transactionChains = new HashMap<>(); + private final DOMTransactionFactory transactionFactory; 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())); - this.name = name; + this.name = name.toString(); this.datastoreContext = datastoreContext; this.schemaContext = schemaContext; this.dataPersistenceProvider = (datastoreContext.isPersistent()) @@ -173,13 +171,17 @@ public class Shard extends RaftActor { 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()); + transactionFactory = new DOMTransactionFactory(store, shardMBean, LOG, this.name); + + commitCoordinator = new ShardCommitCoordinator(transactionFactory, + TimeUnit.SECONDS.convert(5, TimeUnit.MINUTES), + datastoreContext.getShardTransactionCommitQueueCapacity(), self(), LOG, this.name); setTransactionCommitTimeout(); @@ -195,20 +197,8 @@ public class Shard extends RaftActor { 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; - } - 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,6 +262,8 @@ 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); } else if (CanCommitTransaction.SERIALIZABLE_CLASS.isInstance(message)) { @@ -374,9 +366,10 @@ public class Shard extends RaftActor { // currently uses a same thread executor anyway. cohortEntry.getCohort().preCommit().get(); - // If we do not have any followers and we are not using persistence we can - // apply modification to the state immediately - if(!hasFollowers() && !persistence().isRecoveryApplicable()){ + // 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, @@ -450,6 +443,47 @@ public class Shard extends RaftActor { commitCoordinator.handleCanCommit(canCommit, getSender(), self()); } + 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 { + BatchedModificationsReply reply = commitCoordinator.handleTransactionModifications(batched); + sender().tell(reply, self()); + } 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 { + // 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 handleForwardedReadyTransaction(ForwardedReadyTransaction ready) { LOG.debug("{}: Readying transaction {}, client version {}", persistenceId(), ready.getTransactionID(), ready.getTxnClientVersion()); @@ -458,7 +492,7 @@ public class Shard extends RaftActor { // commitCoordinator in preparation for the subsequent three phase commit initiated by // the front-end. commitCoordinator.transactionReady(ready.getTransactionID(), ready.getCohort(), - ready.getModification()); + (MutableCompositeModification) 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 @@ -535,56 +569,18 @@ public class Shard extends RaftActor { } private void closeTransactionChain(final CloseTransactionChain closeTransactionChain) { - DOMStoreTransactionChain chain = - transactionChains.remove(closeTransactionChain.getTransactionChainId()); - - if(chain != null) { - chain.close(); - } + transactionFactory.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(); + DOMStoreTransaction transaction = transactionFactory.newTransaction( + TransactionProxy.TransactionType.fromInt(transactionType), transactionId.toString(), + transactionChainId); - 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); - } + return createShardTransaction(transaction, transactionId, clientVersion); } private ActorRef createShardTransaction(DOMStoreTransaction transaction, ShardTransactionIdentifier transactionId, @@ -792,7 +788,6 @@ public class Shard extends RaftActor { recoveryCoordinator = null; currentLogRecoveryBatch = null; - updateJournalStats(); //notify shard manager getContext().parent().tell(new ActorInitialized(), getSelf()); @@ -831,9 +826,6 @@ public class Shard extends RaftActor { persistenceId(), data, data.getClass().getClassLoader(), CompositeModificationPayload.class.getClassLoader()); } - - updateJournalStats(); - } private void applyModificationToState(ActorRef clientActor, String identifier, Object modification) { @@ -851,19 +843,6 @@ 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 @@ -921,21 +900,15 @@ public class Shard extends RaftActor { delayedListenerRegistrations.clear(); } - shardMBean.setRaftState(getRaftState().name()); - shardMBean.setCurrentTerm(getCurrentTerm()); - // 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(); + transactionFactory.closeAllTransactionChains(); } } @@ -944,12 +917,8 @@ public class Shard extends RaftActor { return dataPersistenceProvider; } - @Override protected void onLeaderChanged(final String oldLeader, final String newLeader) { - shardMBean.setLeader(newLeader); - } - @Override public String persistenceId() { - return this.name.toString(); + return this.name; } @VisibleForTesting @@ -957,16 +926,22 @@ public class Shard extends RaftActor { return dataPersistenceProvider; } + @VisibleForTesting + 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;