X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FShard.java;h=9cd52b219a442320bd0dc8bf12cdfc6b6c4a577b;hb=e2d9f9c57e124d46e117f17c44b77c89222fdb99;hp=66467af1303d9f86e043bd16c5cf604800a72410;hpb=6dcee56392712348b1abdcdc0d1d5f94dfcf505c;p=controller.git 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 66467af130..9cd52b219a 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,19 +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.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; @@ -52,12 +48,11 @@ import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionR 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; @@ -73,13 +68,11 @@ 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; @@ -109,15 +102,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; @@ -140,12 +126,14 @@ public class Shard extends RaftActor { * Coordinates persistence recovery on startup. */ private ShardRecoveryCoordinator recoveryCoordinator; - private List currentLogRecoveryBatch; private final DOMTransactionFactory transactionFactory; 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) { super(name.toString(), new HashMap<>(peerAddresses), Optional.of(datastoreContext.getShardRaftConfig())); @@ -153,18 +141,17 @@ 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); } @@ -190,6 +177,8 @@ public class Shard extends RaftActor { appendEntriesReplyTracker = new MessageTracker(AppendEntriesReply.class, getRaftActorContext().getConfigParams().getIsolatedCheckIntervalInMillis()); + + recoveryCoordinator = new ShardRecoveryCoordinator(store, persistenceId(), LOG); } private void setTransactionCommitTimeout() { @@ -275,7 +264,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 +302,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()); @@ -654,58 +643,7 @@ 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.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(); } @@ -713,81 +651,27 @@ public class Shard extends RaftActor { @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); - } + recoveryCoordinator.startLogRecoveryBatch(maxBatchSize); } @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); - } + recoveryCoordinator.appendRecoveredLogPayload(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()); - } + recoveryCoordinator.applyRecoveredSnapshot(snapshotBytes); } @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()); - } + recoveryCoordinator.applyCurrentLogRecoveryBatch(); } @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; //notify shard manager getContext().parent().tell(new ActorInitialized(), getSelf()); @@ -886,22 +770,11 @@ public class Shard extends RaftActor { @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", @@ -913,19 +786,10 @@ public class Shard extends RaftActor { } @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; @@ -964,45 +828,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(); - } - } - } }