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%2FAbstractDataStore.java;h=05edad2cf1f72867110f5a3798a6570365408682;hb=HEAD;hp=d6958a70d72030ce20b01b8ba30e39149535ab06;hpb=f0e8f9a9547dca1e232a7e8d3912a86de69bcc6a;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractDataStore.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractDataStore.java index d6958a70d7..05edad2cf1 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractDataStore.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractDataStore.java @@ -7,7 +7,6 @@ */ package org.opendaylight.controller.cluster.datastore; -import static com.google.common.base.Preconditions.checkArgument; import static java.util.Objects.requireNonNull; import akka.actor.ActorRef; @@ -17,36 +16,36 @@ import akka.actor.Props; import com.google.common.annotations.Beta; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Throwables; +import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.util.concurrent.SettableFuture; import com.google.common.util.concurrent.Uninterruptibles; -import java.util.Set; -import java.util.concurrent.CountDownLatch; +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; +import java.util.List; +import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import org.eclipse.jdt.annotation.NonNull; import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier; import org.opendaylight.controller.cluster.common.actor.Dispatchers; import org.opendaylight.controller.cluster.databroker.actors.dds.DataStoreClient; import org.opendaylight.controller.cluster.databroker.actors.dds.DistributedDataStoreClientActor; import org.opendaylight.controller.cluster.datastore.config.Configuration; import org.opendaylight.controller.cluster.datastore.identifiers.ShardManagerIdentifier; -import org.opendaylight.controller.cluster.datastore.jmx.mbeans.DatastoreConfigurationMXBeanImpl; -import org.opendaylight.controller.cluster.datastore.jmx.mbeans.DatastoreInfoMXBeanImpl; import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot; import org.opendaylight.controller.cluster.datastore.shardmanager.AbstractShardManagerCreator; import org.opendaylight.controller.cluster.datastore.shardmanager.ShardManagerCreator; import org.opendaylight.controller.cluster.datastore.utils.ActorUtils; -import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils; import org.opendaylight.controller.cluster.datastore.utils.PrimaryShardInfoFutureCache; -import org.opendaylight.mdsal.dom.api.ClusteredDOMDataTreeChangeListener; +import org.opendaylight.mdsal.dom.api.DOMDataBroker.CommitCohortExtension; import org.opendaylight.mdsal.dom.api.DOMDataTreeChangeListener; import org.opendaylight.mdsal.dom.api.DOMDataTreeCommitCohort; -import org.opendaylight.mdsal.dom.api.DOMDataTreeCommitCohortRegistration; -import org.opendaylight.mdsal.dom.api.DOMDataTreeCommitCohortRegistry; import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier; import org.opendaylight.mdsal.dom.spi.store.DOMStoreTreeChangePublisher; -import org.opendaylight.yangtools.concepts.ListenerRegistration; +import org.opendaylight.yangtools.concepts.Registration; +import org.opendaylight.yangtools.yang.common.Empty; import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; +import org.opendaylight.yangtools.yang.data.tree.api.DataTreeCandidate; import org.opendaylight.yangtools.yang.model.api.EffectiveModelContext; -import org.opendaylight.yangtools.yang.model.api.EffectiveModelContextListener; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import scala.concurrent.duration.Duration; @@ -54,13 +53,12 @@ import scala.concurrent.duration.Duration; /** * Base implementation of a distributed DOMStore. */ -public abstract class AbstractDataStore implements DistributedDataStoreInterface, EffectiveModelContextListener, - DatastoreContextPropertiesUpdater.Listener, DOMStoreTreeChangePublisher, - DOMDataTreeCommitCohortRegistry, AutoCloseable { - +public abstract class AbstractDataStore implements DistributedDataStoreInterface, + DatastoreContextPropertiesUpdater.Listener, DOMStoreTreeChangePublisher, CommitCohortExtension, + AutoCloseable { private static final Logger LOG = LoggerFactory.getLogger(AbstractDataStore.class); - private final CountDownLatch waitTillReadyCountDownLatch = new CountDownLatch(1); + private final SettableFuture readinessFuture = SettableFuture.create(); private final ClientIdentifier identifier; private final DataStoreClient client; private final ActorUtils actorUtils; @@ -70,6 +68,7 @@ public abstract class AbstractDataStore implements DistributedDataStoreInterface private DatastoreInfoMXBeanImpl datastoreInfoMXBean; @SuppressWarnings("checkstyle:IllegalCatch") + @SuppressFBWarnings(value = "MC_OVERRIDABLE_METHOD_CALL_IN_CONSTRUCTOR", justification = "Testing overrides") protected AbstractDataStore(final ActorSystem actorSystem, final ClusterWrapper cluster, final Configuration configuration, final DatastoreContextFactory datastoreContextFactory, final DatastoreSnapshot restoreFromSnapshot) { @@ -90,7 +89,7 @@ public abstract class AbstractDataStore implements DistributedDataStoreInterface AbstractShardManagerCreator creator = getShardManagerCreator().cluster(cluster).configuration(configuration) .datastoreContextFactory(datastoreContextFactory) - .waitTillReadyCountDownLatch(waitTillReadyCountDownLatch) + .readinessFuture(readinessFuture) .primaryShardInfoCache(primaryShardInfoCache) .restoreFromSnapshot(restoreFromSnapshot) .distributedDataStore(this); @@ -108,7 +107,7 @@ public abstract class AbstractDataStore implements DistributedDataStoreInterface LOG.error("Failed to get actor for {}", clientProps, e); clientActor.tell(PoisonPill.getInstance(), ActorRef.noSender()); Throwables.throwIfUnchecked(e); - throw new RuntimeException(e); + throw new IllegalStateException(e); } identifier = client.getIdentifier(); @@ -124,21 +123,15 @@ public abstract class AbstractDataStore implements DistributedDataStoreInterface datastoreInfoMXBean.registerMBean(); } - @VisibleForTesting - protected AbstractDataStore(final ActorUtils actorUtils, final ClientIdentifier identifier) { - this.actorUtils = requireNonNull(actorUtils, "actorContext should not be null"); - this.client = null; - this.identifier = requireNonNull(identifier); - } - @VisibleForTesting protected AbstractDataStore(final ActorUtils actorUtils, final ClientIdentifier identifier, final DataStoreClient clientActor) { this.actorUtils = requireNonNull(actorUtils, "actorContext should not be null"); - this.client = clientActor; + client = clientActor; this.identifier = requireNonNull(identifier); } + @VisibleForTesting protected AbstractShardManagerCreator getShardManagerCreator() { return new ShardManagerCreator(); } @@ -147,17 +140,18 @@ public abstract class AbstractDataStore implements DistributedDataStoreInterface return client; } - final ClientIdentifier getIdentifier() { - return identifier; - } - public void setCloseable(final AutoCloseable closeable) { this.closeable = closeable; } @Override - public ListenerRegistration registerTreeChangeListener( - final YangInstanceIdentifier treeId, final L listener) { + public final Registration registerTreeChangeListener(final YangInstanceIdentifier treeId, + final DOMDataTreeChangeListener listener) { + return registerTreeChangeListener(treeId, listener, true); + } + + private @NonNull Registration registerTreeChangeListener(final YangInstanceIdentifier treeId, + final DOMDataTreeChangeListener listener, final boolean clustered) { requireNonNull(treeId, "treeId should not be null"); requireNonNull(listener, "listener should not be null"); @@ -168,47 +162,51 @@ public abstract class AbstractDataStore implements DistributedDataStoreInterface if (treeId.isEmpty()) { // User is targeting root of the datastore. If there is more than one shard, we have to register with them // all and perform data composition. - final Set shardNames = actorUtils.getConfiguration().getAllShardNames(); + final var shardNames = actorUtils.getConfiguration().getAllShardNames(); if (shardNames.size() > 1) { - checkArgument(listener instanceof ClusteredDOMDataTreeChangeListener, - "Cannot listen on root without non-clustered listener %s", listener); + if (!clustered) { + throw new IllegalArgumentException( + "Cannot listen on root without non-clustered listener " + listener); + } return new RootDataTreeChangeListenerProxy<>(actorUtils, listener, shardNames); } } - final String shardName = actorUtils.getShardStrategyFactory().getStrategy(treeId).findShard(treeId); + final var shardName = actorUtils.getShardStrategyFactory().getStrategy(treeId).findShard(treeId); LOG.debug("Registering tree listener: {} for tree: {} shard: {}", listener, treeId, shardName); - final DataTreeChangeListenerProxy listenerRegistrationProxy = - new DataTreeChangeListenerProxy<>(actorUtils, listener, treeId); - listenerRegistrationProxy.init(shardName); + return DataTreeChangeListenerProxy.of(actorUtils, listener, treeId, clustered, shardName); + } - return listenerRegistrationProxy; + @Override + @Deprecated(since = "9.0.0", forRemoval = true) + public final Registration registerLegacyTreeChangeListener(final YangInstanceIdentifier treeId, + final DOMDataTreeChangeListener listener) { + return registerTreeChangeListener(treeId, listener, false); } @Override - public DOMDataTreeCommitCohortRegistration registerCommitCohort( - final DOMDataTreeIdentifier subtree, final C cohort) { - YangInstanceIdentifier treeId = requireNonNull(subtree, "subtree should not be null").getRootIdentifier(); + // Non-final for testing + public Registration registerCommitCohort(final DOMDataTreeIdentifier subtree, + final DOMDataTreeCommitCohort cohort) { + YangInstanceIdentifier treeId = requireNonNull(subtree, "subtree should not be null").path(); requireNonNull(cohort, "listener should not be null"); final String shardName = actorUtils.getShardStrategyFactory().getStrategy(treeId).findShard(treeId); LOG.debug("Registering cohort: {} for tree: {} shard: {}", cohort, treeId, shardName); - DataTreeCohortRegistrationProxy cohortProxy = - new DataTreeCohortRegistrationProxy<>(actorUtils, subtree, cohort); + final var cohortProxy = new DataTreeCohortRegistrationProxy<>(actorUtils, subtree, cohort); cohortProxy.init(shardName); return cohortProxy; } - @Override public void onModelContextUpdated(final EffectiveModelContext newModelContext) { actorUtils.setSchemaContext(newModelContext); } @Override - public void onDatastoreContextUpdated(final DatastoreContextFactory contextFactory) { + public final void onDatastoreContextUpdated(final DatastoreContextFactory contextFactory) { LOG.info("DatastoreContext updated for data store {}", actorUtils.getDataStoreName()); actorUtils.setDatastoreContext(contextFactory); @@ -217,7 +215,7 @@ public abstract class AbstractDataStore implements DistributedDataStoreInterface @Override @SuppressWarnings("checkstyle:IllegalCatch") - public void close() { + public final void close() { LOG.info("Closing data store {}", identifier); if (datastoreConfigMXBean != null) { @@ -243,12 +241,13 @@ public abstract class AbstractDataStore implements DistributedDataStoreInterface } @Override - public ActorUtils getActorUtils() { + public final ActorUtils getActorUtils() { return actorUtils; } // TODO: consider removing this in favor of awaitReadiness() - public void waitTillReady() { + @Deprecated + public final void waitTillReady() { LOG.info("Beginning to wait for data store to become ready : {}", identifier); final Duration toWait = initialSettleTime(); @@ -266,23 +265,37 @@ public abstract class AbstractDataStore implements DistributedDataStoreInterface } @Beta - public boolean awaitReadiness() throws InterruptedException { + @Deprecated + public final boolean awaitReadiness() throws InterruptedException { return awaitReadiness(initialSettleTime()); } @Beta - public boolean awaitReadiness(final Duration toWait) throws InterruptedException { - if (toWait.isFinite()) { - return waitTillReadyCountDownLatch.await(toWait.toNanos(), TimeUnit.NANOSECONDS); + @Deprecated + public final boolean awaitReadiness(final Duration toWait) throws InterruptedException { + try { + if (toWait.isFinite()) { + try { + readinessFuture.get(toWait.toNanos(), TimeUnit.NANOSECONDS); + } catch (TimeoutException e) { + LOG.debug("Timed out waiting for shards to settle", e); + return false; + } + } else { + readinessFuture.get(); + } + } catch (ExecutionException e) { + LOG.warn("Unexpected readiness failure, assuming convergence", e); } - waitTillReadyCountDownLatch.await(); return true; } @Beta - public void awaitReadiness(final long timeout, final TimeUnit unit) throws InterruptedException, TimeoutException { - if (!waitTillReadyCountDownLatch.await(timeout, unit)) { + @Deprecated + public final void awaitReadiness(final long timeout, final TimeUnit unit) + throws InterruptedException, TimeoutException { + if (!awaitReadiness(Duration.create(timeout, unit))) { throw new TimeoutException("Shard leaders failed to settle"); } } @@ -307,48 +320,42 @@ public abstract class AbstractDataStore implements DistributedDataStoreInterface throw new IllegalStateException("Failed to create Shard Manager", lastException); } + /** + * Future which completes when all shards settle for the first time. + * + * @return A Listenable future. + */ + public final ListenableFuture initialSettleFuture() { + return readinessFuture; + } + @VisibleForTesting - public CountDownLatch getWaitTillReadyCountDownLatch() { - return waitTillReadyCountDownLatch; + public final SettableFuture readinessFuture() { + return readinessFuture; } @Override - @SuppressWarnings("unchecked") - public ListenerRegistration registerProxyListener( - final YangInstanceIdentifier shardLookup, final YangInstanceIdentifier insideShard, - final DOMDataTreeChangeListener delegate) { - + public final Registration registerProxyListener(final YangInstanceIdentifier shardLookup, + final YangInstanceIdentifier insideShard, final DOMDataTreeChangeListener delegate) { requireNonNull(shardLookup, "shardLookup should not be null"); requireNonNull(insideShard, "insideShard should not be null"); requireNonNull(delegate, "delegate should not be null"); - final String shardName = actorUtils.getShardStrategyFactory().getStrategy(shardLookup).findShard(shardLookup); - LOG.debug("Registering tree listener: {} for tree: {} shard: {}, path inside shard: {}", - delegate,shardLookup, shardName, insideShard); - - final DataTreeChangeListenerProxy listenerRegistrationProxy = - new DataTreeChangeListenerProxy<>(actorUtils, - // wrap this in the ClusteredDOMDataTreeChangeLister interface - // since we always want clustered registration - (ClusteredDOMDataTreeChangeListener) delegate::onDataTreeChanged, insideShard); - listenerRegistrationProxy.init(shardName); - - return (ListenerRegistration) listenerRegistrationProxy; - } - - @Override - @SuppressWarnings("unchecked") - public ListenerRegistration registerShardConfigListener( - final YangInstanceIdentifier internalPath, final DOMDataTreeChangeListener delegate) { - requireNonNull(delegate, "delegate should not be null"); - - LOG.debug("Registering a listener for the configuration shard: {}", internalPath); + final var shardName = actorUtils.getShardStrategyFactory().getStrategy(shardLookup).findShard(shardLookup); + LOG.debug("Registering tree listener: {} for tree: {} shard: {}, path inside shard: {}", delegate, shardLookup, + shardName, insideShard); - final DataTreeChangeListenerProxy proxy = - new DataTreeChangeListenerProxy<>(actorUtils, delegate, internalPath); - proxy.init(ClusterUtils.PREFIX_CONFIG_SHARD_ID); + return DataTreeChangeListenerProxy.of(actorUtils, new DOMDataTreeChangeListener() { + @Override + public void onDataTreeChanged(final List changes) { + delegate.onDataTreeChanged(changes); + } - return (ListenerRegistration) proxy; + @Override + public void onInitialData() { + delegate.onInitialData(); + } + }, insideShard, true, shardName); } private Duration initialSettleTime() {