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=81449c574780705196e28d3dbd738151c2d948cd;hp=8e00a1389ca6a057bef39292ffcc0b04958be794;hb=c31a6fcf9fb070d4419ca4c32d8b531fdcb5030d;hpb=edcc020c8fda4b13f22a31d79c13feef0b53b0ee 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 8e00a1389c..81449c5747 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,18 +18,15 @@ 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.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; @@ -48,39 +45,31 @@ import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransacti 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.RegisterChangeListener; -import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply; +import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListener; 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.RegisterRoleChangeListener; import org.opendaylight.controller.cluster.notifications.RoleChangeNotifier; import org.opendaylight.controller.cluster.raft.RaftActor; +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.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.model.api.SchemaContext; import scala.concurrent.duration.Duration; import scala.concurrent.duration.FiniteDuration; @@ -93,8 +82,6 @@ import scala.concurrent.duration.FiniteDuration; */ 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 @@ -108,19 +95,8 @@ public class Shard extends RaftActor { 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; @@ -134,15 +110,14 @@ public class Shard extends RaftActor { private final ReadyTransactionReply READY_TRANSACTION_REPLY = new ReadyTransactionReply( Serialization.serializedActorPath(getSelf())); + private final DOMTransactionFactory domTransactionFactory; - /** - * Coordinates persistence recovery on startup. - */ - private ShardRecoveryCoordinator recoveryCoordinator; + private final ShardTransactionActorFactory transactionActorFactory; - private final DOMTransactionFactory transactionFactory; + private final ShardSnapshotCohort snapshotCohort; - private final String txnDispatcherPath; + private final DataTreeChangeListenerSupport treeChangeSupport = new DataTreeChangeListenerSupport(this); + private final DataChangeListenerSupport changeSupport = new DataChangeListenerSupport(this); protected Shard(final ShardIdentifier name, final Map peerAddresses, final DatastoreContext datastoreContext, final SchemaContext schemaContext) { @@ -150,19 +125,15 @@ public class Shard extends RaftActor { 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) { + if (schemaContext != null) { store.onGlobalContextUpdated(schemaContext); } @@ -175,9 +146,9 @@ public class Shard extends RaftActor { getContext().become(new MeteringBehavior(this)); } - transactionFactory = new DOMTransactionFactory(store, shardMBean, LOG, this.name); + domTransactionFactory = new DOMTransactionFactory(store, shardMBean, LOG, this.name); - commitCoordinator = new ShardCommitCoordinator(transactionFactory, + commitCoordinator = new ShardCommitCoordinator(domTransactionFactory, TimeUnit.SECONDS.convert(5, TimeUnit.MINUTES), datastoreContext.getShardTransactionCommitQueueCapacity(), self(), LOG, this.name); @@ -189,7 +160,11 @@ public class Shard extends RaftActor { appendEntriesReplyTracker = new MessageTracker(AppendEntriesReply.class, getRaftActorContext().getConfigParams().getIsolatedCheckIntervalInMillis()); - recoveryCoordinator = new ShardRecoveryCoordinator(store, persistenceId(), LOG); + transactionActorFactory = new ShardTransactionActorFactory(domTransactionFactory, datastoreContext, + new Dispatchers(context().system().dispatchers()).getDispatcherPath( + Dispatchers.DispatcherType.Transaction), self(), getContext(), shardMBean); + + snapshotCohort = new ShardSnapshotCohort(transactionActorFactory, store, LOG, this.name); } private void setTransactionCommitTimeout() { @@ -275,7 +250,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) { @@ -311,12 +288,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()); @@ -458,8 +433,12 @@ public class Shard extends RaftActor { // if(isLeader()) { try { - BatchedModificationsReply reply = commitCoordinator.handleTransactionModifications(batched); - sender().tell(reply, self()); + boolean ready = commitCoordinator.handleTransactionModifications(batched); + if(ready) { + sender().tell(READY_TRANSACTION_REPLY, self()); + } else { + sender().tell(new BatchedModificationsReply(batched.getModifications().size()), self()); + } } catch (Exception e) { LOG.error("{}: Error handling BatchedModifications for Tx {}", persistenceId(), batched.getTransactionID(), e); @@ -499,20 +478,21 @@ public class Shard extends RaftActor { // 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())); + if(ready.getTxnClientVersion() < DataStoreVersions.LITHIUM_VERSION) { + ActorRef replyActorPath = getSelf(); + if(ready.getTxnClientVersion() < DataStoreVersions.HELIUM_1_VERSION) { + LOG.debug("{}: Creating BackwardsCompatibleThreePhaseCommitCohort", persistenceId()); + replyActorPath = getContext().actorOf(BackwardsCompatibleThreePhaseCommitCohort.props( + ready.getTransactionID())); + } ReadyTransactionReply readyTransactionReply = - new ReadyTransactionReply(Serialization.serializedActorPath(replyActorPath)); + new ReadyTransactionReply(Serialization.serializedActorPath(replyActorPath), + ready.getTxnClientVersion()); getSender().tell(ready.isReturnSerialized() ? readyTransactionReply.toSerializable() : - readyTransactionReply, getSelf()); - + readyTransactionReply, getSelf()); } else { - - getSender().tell(ready.isReturnSerialized() ? READY_TRANSACTION_REPLY.toSerializable() : - READY_TRANSACTION_REPLY, getSelf()); + getSender().tell(READY_TRANSACTION_REPLY, getSelf()); } } @@ -569,29 +549,15 @@ public class Shard extends RaftActor { } private void closeTransactionChain(final CloseTransactionChain closeTransactionChain) { - transactionFactory.closeTransactionChain(closeTransactionChain.getTransactionChainId()); + domTransactionFactory.closeTransactionChain(closeTransactionChain.getTransactionChainId()); } private ActorRef createTypedTransactionActor(int transactionType, ShardTransactionIdentifier transactionId, String transactionChainId, short clientVersion ) { - DOMStoreTransaction transaction = transactionFactory.newTransaction( - TransactionProxy.TransactionType.fromInt(transactionType), transactionId.toString(), - transactionChainId); - - return createShardTransaction(transaction, transactionId, clientVersion); - } - - 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(TransactionProxy.TransactionType.fromInt(transactionType), + transactionId, transactionChainId, clientVersion); } private void createTransaction(CreateTransaction createTransaction) { @@ -623,18 +589,11 @@ 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); try { - syncCommitTransaction(tx); + snapshotCohort.syncCommitTransaction(tx); shardMBean.incrementCommittedTransactionCount(); shardMBean.setLastCommittedTransactionTime(System.currentTimeMillis()); } catch (InterruptedException | ExecutionException e) { @@ -644,9 +603,7 @@ public class Shard extends RaftActor { } private void updateSchemaContext(final UpdateSchemaContext message) { - this.schemaContext = message.getSchemaContext(); updateSchemaContext(message.getSchemaContext()); - store.onGlobalContextUpdated(message.getSchemaContext()); } @VisibleForTesting @@ -654,87 +611,24 @@ public class Shard extends RaftActor { store.onGlobalContextUpdated(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), 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) { - recoveryCoordinator.startLogRecoveryBatch(maxBatchSize); - } - - @Override - protected void appendRecoveredLogEntry(final Payload data) { - recoveryCoordinator.appendRecoveredLogPayload(data); + protected RaftActorSnapshotCohort getRaftActorSnapshotCohort() { + return snapshotCohort; } @Override - protected void applyRecoverySnapshot(final byte[] snapshotBytes) { - recoveryCoordinator.applyRecoveredSnapshot(snapshotBytes); - } - - @Override - protected void applyCurrentLogRecoveryBatch() { - recoveryCoordinator.applyCurrentLogRecoveryBatch(); + @Nonnull + protected RaftActorRecoveryCohort getRaftActorRecoveryCohort() { + return new ShardRecoveryCoordinator(store, persistenceId(), LOG); } @Override protected void onRecoveryComplete() { - recoveryCoordinator = null; - //notify shard manager getContext().parent().tell(new ActorInitialized(), getSelf()); @@ -789,89 +683,29 @@ public class Shard extends RaftActor { } } - @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(); - } + changeSupport.onLeadershipChange(isLeader); + treeChangeSupport.onLeadershipChange(isLeader); // If this actor is no longer the leader close all the transaction chains - if(!isLeader) { + if (!isLeader) { if(LOG.isDebugEnabled()) { LOG.debug( "{}: onStateChanged: Closing all transaction chains because shard {} is no longer the leader", persistenceId(), getId()); } - transactionFactory.closeAllTransactionChains(); + domTransactionFactory.closeAllTransactionChains(); } } @Override - protected DataPersistenceProvider persistence() { - return dataPersistenceProvider; - } - - @Override public String persistenceId() { + public String persistenceId() { return this.name; } - @VisibleForTesting - DataPersistenceProvider getDataPersistenceProvider() { - return dataPersistenceProvider; - } - @VisibleForTesting ShardCommitCoordinator getCommitCoordinator() { return commitCoordinator; @@ -910,45 +744,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(); - } - } - } }