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=17f1abb92cc4ec7d4d0a7dffb818cec20b5cff0b;hb=5de57714fa057ac80f930c2fcce2758ca0a5f514;hp=c04256a28efb5b01dd94429f81cb72b71ae3e1eb;hpb=faad081396a7c6489d234809ca9446ab96d14504;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 c04256a28e..17f1abb92c 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,13 +18,11 @@ 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; @@ -50,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; @@ -66,18 +63,16 @@ 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.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; @@ -107,11 +102,6 @@ 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 SchemaContext schemaContext; @@ -131,16 +121,13 @@ public class Shard extends RaftActor { private final ReadyTransactionReply READY_TRANSACTION_REPLY = new ReadyTransactionReply( Serialization.serializedActorPath(getSelf())); - - /** - * Coordinates persistence recovery on startup. - */ - private ShardRecoveryCoordinator recoveryCoordinator; - 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())); @@ -158,7 +145,7 @@ public class Shard extends RaftActor { store = InMemoryDOMDataStoreFactory.create(name.toString(), null, datastoreContext.getDataStoreProperties()); - if(schemaContext != null) { + if (schemaContext != null) { store.onGlobalContextUpdated(schemaContext); } @@ -184,8 +171,6 @@ public class Shard extends RaftActor { appendEntriesReplyTracker = new MessageTracker(AppendEntriesReply.class, getRaftActorContext().getConfigParams().getIsolatedCheckIntervalInMillis()); - - recoveryCoordinator = new ShardRecoveryCoordinator(store, persistenceId(), LOG); } private void setTransactionCommitTimeout() { @@ -271,7 +256,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) { @@ -648,87 +635,19 @@ 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); - } - - @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()); @@ -826,22 +745,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", @@ -895,45 +803,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(); - } - } - } }