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%2FAbstractDataStore.java;h=70fef4197e3c4a8f4faa71e59753a57ef70708cc;hp=bc4a8e739577d6264038594d67980673fc937fe2;hb=d33f2a4667e423fcbff25454f28412a5b3d01a42;hpb=b65f82f3c6e6e4c53ea2efa4dba30ddf82d61616 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 bc4a8e7395..70fef4197e 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,6 +7,7 @@ */ package org.opendaylight.controller.cluster.datastore; +import static com.google.common.base.Preconditions.checkArgument; import static java.util.Objects.requireNonNull; import akka.actor.ActorRef; @@ -16,8 +17,12 @@ 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.concurrent.CountDownLatch; +import java.util.Collection; +import java.util.Set; +import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier; @@ -26,13 +31,10 @@ 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.DOMDataTreeChangeListener; @@ -43,8 +45,9 @@ 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.yang.data.api.YangInstanceIdentifier; -import org.opendaylight.yangtools.yang.model.api.SchemaContext; -import org.opendaylight.yangtools.yang.model.api.SchemaContextListener; +import org.opendaylight.yangtools.yang.data.api.schema.tree.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; @@ -52,13 +55,13 @@ import scala.concurrent.duration.Duration; /** * Base implementation of a distributed DOMStore. */ -public abstract class AbstractDataStore implements DistributedDataStoreInterface, SchemaContextListener, +public abstract class AbstractDataStore implements DistributedDataStoreInterface, EffectiveModelContextListener, DatastoreContextPropertiesUpdater.Listener, DOMStoreTreeChangePublisher, DOMDataTreeCommitCohortRegistry, 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; @@ -88,7 +91,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); @@ -159,6 +162,21 @@ public abstract class AbstractDataStore implements DistributedDataStoreInterface requireNonNull(treeId, "treeId should not be null"); requireNonNull(listener, "listener should not be null"); + /* + * We need to potentially deal with multi-shard composition for registration targeting the root of the data + * store. If that is the case, we delegate to a more complicated setup invol + */ + 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(); + if (shardNames.size() > 1) { + checkArgument(listener instanceof ClusteredDOMDataTreeChangeListener, + "Cannot listen on root without non-clustered listener %s", listener); + return new RootDataTreeChangeListenerProxy<>(actorUtils, listener, shardNames); + } + } + final String shardName = actorUtils.getShardStrategyFactory().getStrategy(treeId).findShard(treeId); LOG.debug("Registering tree listener: {} for tree: {} shard: {}", listener, treeId, shardName); @@ -169,7 +187,6 @@ public abstract class AbstractDataStore implements DistributedDataStoreInterface return listenerRegistrationProxy; } - @Override public DOMDataTreeCommitCohortRegistration registerCommitCohort( final DOMDataTreeIdentifier subtree, final C cohort) { @@ -187,8 +204,8 @@ public abstract class AbstractDataStore implements DistributedDataStoreInterface } @Override - public void onGlobalContextUpdated(final SchemaContext schemaContext) { - actorUtils.setSchemaContext(schemaContext); + public void onModelContextUpdated(final EffectiveModelContext newModelContext) { + actorUtils.setSchemaContext(newModelContext); } @Override @@ -232,18 +249,15 @@ public abstract class AbstractDataStore implements DistributedDataStoreInterface } // TODO: consider removing this in favor of awaitReadiness() + @Deprecated public void waitTillReady() { LOG.info("Beginning to wait for data store to become ready : {}", identifier); final Duration toWait = initialSettleTime(); try { - if (toWait.isFinite()) { - if (!waitTillReadyCountDownLatch.await(toWait.toNanos(), TimeUnit.NANOSECONDS)) { - LOG.error("Shard leaders failed to settle in {}, giving up", toWait); - return; - } - } else { - waitTillReadyCountDownLatch.await(); + if (!awaitReadiness(toWait)) { + LOG.error("Shard leaders failed to settle in {}, giving up", toWait); + return; } } catch (InterruptedException e) { LOG.error("Interrupted while waiting for shards to settle", e); @@ -254,8 +268,36 @@ public abstract class AbstractDataStore implements DistributedDataStoreInterface } @Beta + @Deprecated + public boolean awaitReadiness() throws InterruptedException { + return awaitReadiness(initialSettleTime()); + } + + @Beta + @Deprecated + public 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); + } + + return true; + } + + @Beta + @Deprecated public void awaitReadiness(final long timeout, final TimeUnit unit) throws InterruptedException, TimeoutException { - if (!waitTillReadyCountDownLatch.await(timeout, unit)) { + if (!awaitReadiness(Duration.create(timeout, unit))) { throw new TimeoutException("Shard leaders failed to settle"); } } @@ -280,16 +322,25 @@ 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; + SettableFuture readinessFuture() { + return readinessFuture; } + @Override @SuppressWarnings("unchecked") public ListenerRegistration registerProxyListener( - final YangInstanceIdentifier shardLookup, - final YangInstanceIdentifier insideShard, - final org.opendaylight.mdsal.dom.api.DOMDataTreeChangeListener delegate) { + final YangInstanceIdentifier shardLookup, final YangInstanceIdentifier insideShard, + final DOMDataTreeChangeListener delegate) { requireNonNull(shardLookup, "shardLookup should not be null"); requireNonNull(insideShard, "insideShard should not be null"); @@ -299,31 +350,25 @@ public abstract class AbstractDataStore implements DistributedDataStoreInterface LOG.debug("Registering tree listener: {} for tree: {} shard: {}, path inside shard: {}", delegate,shardLookup, shardName, insideShard); + // wrap this in the ClusteredDOMDataTreeChangeLister interface + // since we always want clustered registration final DataTreeChangeListenerProxy listenerRegistrationProxy = - new DataTreeChangeListenerProxy<>(actorUtils, - // wrap this in the ClusteredDOMDataTreeChangeLister interface - // since we always want clustered registration - (ClusteredDOMDataTreeChangeListener) delegate::onDataTreeChanged, insideShard); + new DataTreeChangeListenerProxy<>(actorUtils, new ClusteredDOMDataTreeChangeListener() { + @Override + public void onDataTreeChanged(final Collection changes) { + delegate.onDataTreeChanged(changes); + } + + @Override + public void onInitialData() { + delegate.onInitialData(); + } + }, insideShard); listenerRegistrationProxy.init(shardName); return (ListenerRegistration) listenerRegistrationProxy; } - @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 DataTreeChangeListenerProxy proxy = - new DataTreeChangeListenerProxy<>(actorUtils, delegate, internalPath); - proxy.init(ClusterUtils.PREFIX_CONFIG_SHARD_ID); - - return (ListenerRegistration) proxy; - } - private Duration initialSettleTime() { final DatastoreContext context = actorUtils.getDatastoreContext(); final int multiplier = context.getInitialSettleTimeoutMultiplier();