X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;ds=sidebyside;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FAbstractDataStore.java;h=8537c85698fa6a2759c224da3a7488d8c24dd95d;hb=ccca30bbb1545643c427fc59c23329c5d49f8d4b;hp=43314cd3ba0c7ab84987ea2a695533f852bf850a;hpb=3115b8171461584e85f58d87a9f179013cfbb262;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 43314cd3ba..8537c85698 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,17 +7,21 @@ */ package org.opendaylight.controller.cluster.datastore; +import static com.google.common.base.Preconditions.checkArgument; import static java.util.Objects.requireNonNull; import akka.actor.ActorRef; import akka.actor.ActorSystem; import akka.actor.PoisonPill; 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.Uninterruptibles; +import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; 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; @@ -41,36 +45,30 @@ 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.model.api.EffectiveModelContext; +import org.opendaylight.yangtools.yang.model.api.EffectiveModelContextListener; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +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 static final long READY_WAIT_FACTOR = 3; - + private final CountDownLatch waitTillReadyCountDownLatch = new CountDownLatch(1); + private final ClientIdentifier identifier; + private final DataStoreClient client; private final ActorUtils actorUtils; - private final long waitTillReadyTimeInMillis; private AutoCloseable closeable; - private DatastoreConfigurationMXBeanImpl datastoreConfigMXBean; - private DatastoreInfoMXBeanImpl datastoreInfoMXBean; - private final CountDownLatch waitTillReadyCountDownLatch = new CountDownLatch(1); - - private final ClientIdentifier identifier; - private final DataStoreClient client; - @SuppressWarnings("checkstyle:IllegalCatch") protected AbstractDataStore(final ActorSystem actorSystem, final ClusterWrapper cluster, final Configuration configuration, final DatastoreContextFactory datastoreContextFactory, @@ -116,9 +114,6 @@ public abstract class AbstractDataStore implements DistributedDataStoreInterface identifier = client.getIdentifier(); LOG.debug("Distributed data store client {} started", identifier); - this.waitTillReadyTimeInMillis = actorUtils.getDatastoreContext().getShardLeaderElectionTimeout() - .duration().toMillis() * READY_WAIT_FACTOR; - datastoreConfigMXBean = new DatastoreConfigurationMXBeanImpl( datastoreContextFactory.getBaseDatastoreContext().getDataStoreMXBeanType()); datastoreConfigMXBean.setContext(datastoreContextFactory.getBaseDatastoreContext()); @@ -134,8 +129,6 @@ public abstract class AbstractDataStore implements DistributedDataStoreInterface this.actorUtils = requireNonNull(actorUtils, "actorContext should not be null"); this.client = null; this.identifier = requireNonNull(identifier); - this.waitTillReadyTimeInMillis = actorUtils.getDatastoreContext().getShardLeaderElectionTimeout() - .duration().toMillis() * READY_WAIT_FACTOR; } @VisibleForTesting @@ -144,8 +137,6 @@ public abstract class AbstractDataStore implements DistributedDataStoreInterface this.actorUtils = requireNonNull(actorUtils, "actorContext should not be null"); this.client = clientActor; this.identifier = requireNonNull(identifier); - this.waitTillReadyTimeInMillis = actorUtils.getDatastoreContext().getShardLeaderElectionTimeout() - .duration().toMillis() * READY_WAIT_FACTOR; } protected AbstractShardManagerCreator getShardManagerCreator() { @@ -170,6 +161,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); @@ -180,7 +186,6 @@ public abstract class AbstractDataStore implements DistributedDataStoreInterface return listenerRegistrationProxy; } - @Override public DOMDataTreeCommitCohortRegistration registerCommitCohort( final DOMDataTreeIdentifier subtree, final C cohort) { @@ -198,8 +203,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 @@ -242,18 +247,32 @@ public abstract class AbstractDataStore implements DistributedDataStoreInterface return actorUtils; } + // TODO: consider removing this in favor of awaitReadiness() public void waitTillReady() { LOG.info("Beginning to wait for data store to become ready : {}", identifier); + final Duration toWait = initialSettleTime(); try { - if (waitTillReadyCountDownLatch.await(waitTillReadyTimeInMillis, TimeUnit.MILLISECONDS)) { - LOG.debug("Data store {} is now ready", identifier); + if (toWait.isFinite()) { + if (!waitTillReadyCountDownLatch.await(toWait.toNanos(), TimeUnit.NANOSECONDS)) { + LOG.error("Shard leaders failed to settle in {}, giving up", toWait); + return; + } } else { - LOG.error("Shard leaders failed to settle in {} seconds, giving up", - TimeUnit.MILLISECONDS.toSeconds(waitTillReadyTimeInMillis)); + waitTillReadyCountDownLatch.await(); } } catch (InterruptedException e) { LOG.error("Interrupted while waiting for shards to settle", e); + return; + } + + LOG.debug("Data store {} is now ready", identifier); + } + + @Beta + public void awaitReadiness(final long timeout, final TimeUnit unit) throws InterruptedException, TimeoutException { + if (!waitTillReadyCountDownLatch.await(timeout, unit)) { + throw new TimeoutException("Shard leaders failed to settle"); } } @@ -284,9 +303,8 @@ public abstract class AbstractDataStore implements DistributedDataStoreInterface @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"); @@ -321,4 +339,9 @@ public abstract class AbstractDataStore implements DistributedDataStoreInterface return (ListenerRegistration) proxy; } + private Duration initialSettleTime() { + final DatastoreContext context = actorUtils.getDatastoreContext(); + final int multiplier = context.getInitialSettleTimeoutMultiplier(); + return multiplier == 0 ? Duration.Inf() : context.getShardLeaderElectionTimeout().duration().$times(multiplier); + } }