*/
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;
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,
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());
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
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() {
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<String> 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);
return listenerRegistrationProxy;
}
-
@Override
public <C extends DOMDataTreeCommitCohort> DOMDataTreeCommitCohortRegistration<C> registerCommitCohort(
final DOMDataTreeIdentifier subtree, final C cohort) {
}
@Override
- public void onGlobalContextUpdated(final SchemaContext schemaContext) {
- actorUtils.setSchemaContext(schemaContext);
+ public void onModelContextUpdated(final EffectiveModelContext newModelContext) {
+ actorUtils.setSchemaContext(newModelContext);
}
@Override
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);
- } else {
- LOG.error("Shard leaders failed to settle in {} seconds, giving up",
- TimeUnit.MILLISECONDS.toSeconds(waitTillReadyTimeInMillis));
+ 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);
+ return;
+ }
+
+ LOG.debug("Data store {} is now ready", identifier);
+ }
+
+ @Beta
+ public 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);
+ }
+
+ waitTillReadyCountDownLatch.await();
+ return true;
+ }
+
+ @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");
}
}
return waitTillReadyCountDownLatch;
}
+ @Override
@SuppressWarnings("unchecked")
public <L extends DOMDataTreeChangeListener> ListenerRegistration<L> 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");
return (ListenerRegistration<L>) listenerRegistrationProxy;
}
+ @Override
@SuppressWarnings("unchecked")
public <L extends DOMDataTreeChangeListener> ListenerRegistration<L> registerShardConfigListener(
- final YangInstanceIdentifier internalPath,
- final DOMDataTreeChangeListener delegate) {
+ final YangInstanceIdentifier internalPath, final DOMDataTreeChangeListener delegate) {
requireNonNull(delegate, "delegate should not be null");
LOG.debug("Registering a listener for the configuration shard: {}", internalPath);
return (ListenerRegistration<L>) proxy;
}
+ private Duration initialSettleTime() {
+ final DatastoreContext context = actorUtils.getDatastoreContext();
+ final int multiplier = context.getInitialSettleTimeoutMultiplier();
+ return multiplier == 0 ? Duration.Inf() : context.getShardLeaderElectionTimeout().duration().$times(multiplier);
+ }
}
Code Review / controller.git / blobdiff