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%2Futils%2FActorContext.java;h=62652342ded45979b09cc647a76be42f8347a946;hb=refs%2Fchanges%2F49%2F78449%2F2;hp=5c514cf77502587fc15939df20e61187d0dfe297;hpb=d3a97264ecf47e8c60ea11a7caebce41b580e91d;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/utils/ActorContext.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/utils/ActorContext.java index 5c514cf775..62652342de 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/utils/ActorContext.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/utils/ActorContext.java @@ -5,19 +5,19 @@ * terms of the Eclipse Public License v1.0 which accompanies this distribution, * and is available at http://www.eclipse.org/legal/epl-v10.html */ - package org.opendaylight.controller.cluster.datastore.utils; import static akka.pattern.Patterns.ask; + import akka.actor.ActorPath; import akka.actor.ActorRef; import akka.actor.ActorSelection; import akka.actor.ActorSystem; import akka.actor.Address; -import akka.actor.PoisonPill; import akka.dispatch.Mapper; import akka.dispatch.OnComplete; import akka.pattern.AskTimeoutException; +import akka.pattern.Patterns; import akka.util.Timeout; import com.codahale.metrics.MetricRegistry; import com.codahale.metrics.Timer; @@ -25,10 +25,14 @@ import com.google.common.base.Optional; import com.google.common.base.Preconditions; import com.google.common.base.Strings; import java.util.concurrent.TimeUnit; -import org.opendaylight.controller.cluster.common.actor.CommonConfig; +import java.util.function.Function; +import org.opendaylight.controller.cluster.access.concepts.MemberName; +import org.opendaylight.controller.cluster.common.actor.Dispatchers; import org.opendaylight.controller.cluster.datastore.ClusterWrapper; -import org.opendaylight.controller.cluster.datastore.Configuration; +import org.opendaylight.controller.cluster.datastore.DataStoreVersions; import org.opendaylight.controller.cluster.datastore.DatastoreContext; +import org.opendaylight.controller.cluster.datastore.DatastoreContextFactory; +import org.opendaylight.controller.cluster.datastore.config.Configuration; import org.opendaylight.controller.cluster.datastore.exceptions.LocalShardNotFoundException; import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException; import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedException; @@ -43,7 +47,10 @@ import org.opendaylight.controller.cluster.datastore.messages.LocalShardNotFound import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo; import org.opendaylight.controller.cluster.datastore.messages.RemotePrimaryShardFound; import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext; +import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory; +import org.opendaylight.controller.cluster.raft.client.messages.Shutdown; import org.opendaylight.controller.cluster.reporting.MetricsReporter; +import org.opendaylight.mdsal.common.api.LogicalDatastoreType; import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree; import org.opendaylight.yangtools.yang.model.api.SchemaContext; import org.slf4j.Logger; @@ -51,7 +58,6 @@ import org.slf4j.LoggerFactory; import scala.concurrent.Await; import scala.concurrent.ExecutionContext; import scala.concurrent.Future; -import scala.concurrent.duration.Duration; import scala.concurrent.duration.FiniteDuration; /** @@ -67,19 +73,19 @@ public class ActorContext { private static final Mapper FIND_PRIMARY_FAILURE_TRANSFORMER = new Mapper() { @Override - public Throwable apply(Throwable failure) { + public Throwable apply(final Throwable failure) { Throwable actualFailure = failure; - if(failure instanceof AskTimeoutException) { + if (failure instanceof AskTimeoutException) { // A timeout exception most likely means the shard isn't initialized. actualFailure = new NotInitializedException( - "Timed out trying to find the primary shard. Most likely cause is the " + - "shard is not initialized yet."); + "Timed out trying to find the primary shard. Most likely cause is the " + + "shard is not initialized yet."); } return actualFailure; } }; - public static final String MAILBOX = "bounded-mailbox"; + public static final String BOUNDED_MAILBOX = "bounded-mailbox"; public static final String COMMIT = "commit"; private final ActorSystem actorSystem; @@ -91,26 +97,31 @@ public class ActorContext { private Timeout operationTimeout; private final String selfAddressHostPort; private TransactionRateLimiter txRateLimiter; - private final int transactionOutstandingOperationLimit; private Timeout transactionCommitOperationTimeout; private Timeout shardInitializationTimeout; private final Dispatchers dispatchers; private volatile SchemaContext schemaContext; + + // Used as a write memory barrier. + @SuppressWarnings("unused") private volatile boolean updated; - private final MetricRegistry metricRegistry = MetricsReporter.getInstance(DatastoreContext.METRICS_DOMAIN).getMetricsRegistry(); + + private final MetricRegistry metricRegistry = MetricsReporter.getInstance(DatastoreContext.METRICS_DOMAIN) + .getMetricsRegistry(); private final PrimaryShardInfoFutureCache primaryShardInfoCache; + private final ShardStrategyFactory shardStrategyFactory; - public ActorContext(ActorSystem actorSystem, ActorRef shardManager, - ClusterWrapper clusterWrapper, Configuration configuration) { + public ActorContext(final ActorSystem actorSystem, final ActorRef shardManager, + final ClusterWrapper clusterWrapper, final Configuration configuration) { this(actorSystem, shardManager, clusterWrapper, configuration, DatastoreContext.newBuilder().build(), new PrimaryShardInfoFutureCache()); } - public ActorContext(ActorSystem actorSystem, ActorRef shardManager, - ClusterWrapper clusterWrapper, Configuration configuration, - DatastoreContext datastoreContext, PrimaryShardInfoFutureCache primaryShardInfoCache) { + public ActorContext(final ActorSystem actorSystem, final ActorRef shardManager, + final ClusterWrapper clusterWrapper, final Configuration configuration, + final DatastoreContext datastoreContext, final PrimaryShardInfoFutureCache primaryShardInfoCache) { this.actorSystem = actorSystem; this.shardManager = shardManager; this.clusterWrapper = clusterWrapper; @@ -119,6 +130,10 @@ public class ActorContext { this.dispatchers = new Dispatchers(actorSystem.dispatchers()); this.primaryShardInfoCache = primaryShardInfoCache; + final LogicalDatastoreType convertedType = + LogicalDatastoreType.valueOf(datastoreContext.getLogicalStoreType().name()); + this.shardStrategyFactory = new ShardStrategyFactory(configuration, convertedType); + setCachedProperties(); Address selfAddress = clusterWrapper.getSelfAddress(); @@ -128,16 +143,16 @@ public class ActorContext { selfAddressHostPort = null; } - transactionOutstandingOperationLimit = new CommonConfig(this.getActorSystem().settings().config()).getMailBoxCapacity(); } private void setCachedProperties() { txRateLimiter = new TransactionRateLimiter(this); - operationDuration = Duration.create(datastoreContext.getOperationTimeoutInSeconds(), TimeUnit.SECONDS); + operationDuration = FiniteDuration.create(datastoreContext.getOperationTimeoutInMillis(), + TimeUnit.MILLISECONDS); operationTimeout = new Timeout(operationDuration); - transactionCommitOperationTimeout = new Timeout(Duration.create( + transactionCommitOperationTimeout = new Timeout(FiniteDuration.create( datastoreContext.getShardTransactionCommitTimeoutInSeconds(), TimeUnit.SECONDS)); shardInitializationTimeout = new Timeout(datastoreContext.getShardInitializationTimeout().duration().$times(2)); @@ -155,24 +170,24 @@ public class ActorContext { return shardManager; } - public ActorSelection actorSelection(String actorPath) { + public ActorSelection actorSelection(final String actorPath) { return actorSystem.actorSelection(actorPath); } - public ActorSelection actorSelection(ActorPath actorPath) { + public ActorSelection actorSelection(final ActorPath actorPath) { return actorSystem.actorSelection(actorPath); } - public void setSchemaContext(SchemaContext schemaContext) { + public void setSchemaContext(final SchemaContext schemaContext) { this.schemaContext = schemaContext; - if(shardManager != null) { + if (shardManager != null) { shardManager.tell(new UpdateSchemaContext(schemaContext), ActorRef.noSender()); } } - public void setDatastoreContext(DatastoreContext context) { - this.datastoreContext = context; + public void setDatastoreContext(final DatastoreContextFactory contextFactory) { + this.datastoreContext = contextFactory.getBaseDatastoreContext(); setCachedProperties(); // We write the 'updated' volatile to trigger a write memory barrier so that the writes above @@ -184,8 +199,8 @@ public class ActorContext { updated = true; - if(shardManager != null) { - shardManager.tell(context, ActorRef.noSender()); + if (shardManager != null) { + shardManager.tell(contextFactory, ActorRef.noSender()); } } @@ -195,7 +210,7 @@ public class ActorContext { public Future findPrimaryShardAsync(final String shardName) { Future ret = primaryShardInfoCache.getIfPresent(shardName); - if(ret != null){ + if (ret != null) { return ret; } Future future = executeOperationAsync(shardManager, @@ -203,19 +218,21 @@ public class ActorContext { return future.transform(new Mapper() { @Override - public PrimaryShardInfo checkedApply(Object response) throws Exception { - if(response instanceof RemotePrimaryShardFound) { + public PrimaryShardInfo checkedApply(final Object response) throws UnknownMessageException { + if (response instanceof RemotePrimaryShardFound) { LOG.debug("findPrimaryShardAsync received: {}", response); - return onPrimaryShardFound(shardName, ((RemotePrimaryShardFound)response).getPrimaryPath(), null); - } else if(response instanceof LocalPrimaryShardFound) { + RemotePrimaryShardFound found = (RemotePrimaryShardFound)response; + return onPrimaryShardFound(shardName, found.getPrimaryPath(), found.getPrimaryVersion(), null); + } else if (response instanceof LocalPrimaryShardFound) { LOG.debug("findPrimaryShardAsync received: {}", response); LocalPrimaryShardFound found = (LocalPrimaryShardFound)response; - return onPrimaryShardFound(shardName, found.getPrimaryPath(), found.getLocalShardDataTree()); - } else if(response instanceof NotInitializedException) { + return onPrimaryShardFound(shardName, found.getPrimaryPath(), DataStoreVersions.CURRENT_VERSION, + found.getLocalShardDataTree()); + } else if (response instanceof NotInitializedException) { throw (NotInitializedException)response; - } else if(response instanceof PrimaryNotFoundException) { + } else if (response instanceof PrimaryNotFoundException) { throw (PrimaryNotFoundException)response; - } else if(response instanceof NoShardLeaderException) { + } else if (response instanceof NoShardLeaderException) { throw (NoShardLeaderException)response; } @@ -225,22 +242,23 @@ public class ActorContext { }, FIND_PRIMARY_FAILURE_TRANSFORMER, getClientDispatcher()); } - private PrimaryShardInfo onPrimaryShardFound(String shardName, String primaryActorPath, - DataTree localShardDataTree) { + private PrimaryShardInfo onPrimaryShardFound(final String shardName, final String primaryActorPath, + final short primaryVersion, final DataTree localShardDataTree) { ActorSelection actorSelection = actorSystem.actorSelection(primaryActorPath); - PrimaryShardInfo info = new PrimaryShardInfo(actorSelection, Optional.fromNullable(localShardDataTree)); + PrimaryShardInfo info = localShardDataTree == null ? new PrimaryShardInfo(actorSelection, primaryVersion) : + new PrimaryShardInfo(actorSelection, primaryVersion, localShardDataTree); primaryShardInfoCache.putSuccessful(shardName, info); return info; } /** - * Finds a local shard given its shard name and return it's ActorRef + * Finds a local shard given its shard name and return it's ActorRef. * * @param shardName the name of the local shard that needs to be found * @return a reference to a local shard actor which represents the shard * specified by the shardName */ - public Optional findLocalShard(String shardName) { + public Optional findLocalShard(final String shardName) { Object result = executeOperation(shardManager, new FindLocalShard(shardName, false)); if (result instanceof LocalShardFound) { @@ -258,20 +276,20 @@ public class ActorContext { * * @param shardName the name of the local shard that needs to be found */ - public Future findLocalShardAsync( final String shardName) { + public Future findLocalShardAsync(final String shardName) { Future future = executeOperationAsync(shardManager, new FindLocalShard(shardName, true), shardInitializationTimeout); return future.map(new Mapper() { @Override - public ActorRef checkedApply(Object response) throws Throwable { - if(response instanceof LocalShardFound) { + public ActorRef checkedApply(final Object response) throws Throwable { + if (response instanceof LocalShardFound) { LocalShardFound found = (LocalShardFound)response; LOG.debug("Local shard found {}", found.getPath()); return found.getPath(); - } else if(response instanceof NotInitializedException) { + } else if (response instanceof NotInitializedException) { throw (NotInitializedException)response; - } else if(response instanceof LocalShardNotFound) { + } else if (response instanceof LocalShardNotFound) { throw new LocalShardNotFoundException( String.format("Local shard for %s does not exist.", shardName)); } @@ -283,49 +301,51 @@ public class ActorContext { } /** - * Executes an operation on a local actor and wait for it's response + * Executes an operation on a local actor and wait for it's response. * - * @param actor - * @param message + * @param actor the actor + * @param message the message to send * @return The response of the operation */ - public Object executeOperation(ActorRef actor, Object message) { + @SuppressWarnings("checkstyle:IllegalCatch") + public Object executeOperation(final ActorRef actor, final Object message) { Future future = executeOperationAsync(actor, message, operationTimeout); try { return Await.result(future, operationDuration); } catch (Exception e) { - throw new TimeoutException("Sending message " + message.getClass().toString() + - " to actor " + actor.toString() + " failed. Try again later.", e); + throw new TimeoutException("Sending message " + message.getClass().toString() + + " to actor " + actor.toString() + " failed. Try again later.", e); } } - public Future executeOperationAsync(ActorRef actor, Object message, Timeout timeout) { - Preconditions.checkArgument(actor != null, "actor must not be null"); - Preconditions.checkArgument(message != null, "message must not be null"); - - LOG.debug("Sending message {} to {}", message.getClass(), actor); - return doAsk(actor, message, timeout); - } - /** - * Execute an operation on a remote actor and wait for it's response + * Execute an operation on a remote actor and wait for it's response. * - * @param actor - * @param message - * @return + * @param actor the actor + * @param message the message + * @return the response message */ - public Object executeOperation(ActorSelection actor, Object message) { + @SuppressWarnings("checkstyle:IllegalCatch") + public Object executeOperation(final ActorSelection actor, final Object message) { Future future = executeOperationAsync(actor, message); try { return Await.result(future, operationDuration); } catch (Exception e) { - throw new TimeoutException("Sending message " + message.getClass().toString() + - " to actor " + actor.toString() + " failed. Try again later.", e); + throw new TimeoutException("Sending message " + message.getClass().toString() + + " to actor " + actor.toString() + " failed. Try again later.", e); } } + public Future executeOperationAsync(final ActorRef actor, final Object message, final Timeout timeout) { + Preconditions.checkArgument(actor != null, "actor must not be null"); + Preconditions.checkArgument(message != null, "message must not be null"); + + LOG.debug("Sending message {} to {}", message.getClass(), actor); + return doAsk(actor, message, timeout); + } + /** * Execute an operation on a remote actor asynchronously. * @@ -334,8 +354,8 @@ public class ActorContext { * @param timeout the operation timeout * @return a Future containing the eventual result */ - public Future executeOperationAsync(ActorSelection actor, Object message, - Timeout timeout) { + public Future executeOperationAsync(final ActorSelection actor, final Object message, + final Timeout timeout) { Preconditions.checkArgument(actor != null, "actor must not be null"); Preconditions.checkArgument(message != null, "message must not be null"); @@ -351,7 +371,7 @@ public class ActorContext { * @param message the message to send * @return a Future containing the eventual result */ - public Future executeOperationAsync(ActorSelection actor, Object message) { + public Future executeOperationAsync(final ActorSelection actor, final Object message) { return executeOperationAsync(actor, message, operationTimeout); } @@ -362,7 +382,7 @@ public class ActorContext { * @param actor the ActorSelection * @param message the message to send */ - public void sendOperationAsync(ActorSelection actor, Object message) { + public void sendOperationAsync(final ActorSelection actor, final Object message) { Preconditions.checkArgument(actor != null, "actor must not be null"); Preconditions.checkArgument(message != null, "message must not be null"); @@ -371,34 +391,39 @@ public class ActorContext { actor.tell(message, ActorRef.noSender()); } + @SuppressWarnings("checkstyle:IllegalCatch") public void shutdown() { - shardManager.tell(PoisonPill.getInstance(), ActorRef.noSender()); + FiniteDuration duration = datastoreContext.getShardRaftConfig().getElectionTimeOutInterval().$times(3); + try { + Await.ready(Patterns.gracefulStop(shardManager, duration, Shutdown.INSTANCE), duration); + } catch (Exception e) { + LOG.warn("ShardManager for {} data store did not shutdown gracefully", getDataStoreName(), e); + } } public ClusterWrapper getClusterWrapper() { return clusterWrapper; } - public String getCurrentMemberName(){ + public MemberName getCurrentMemberName() { return clusterWrapper.getCurrentMemberName(); } /** - * Send the message to each and every shard - * - * @param message + * Send the message to each and every shard. */ - public void broadcast(final Object message){ - for(final String shardName : configuration.getAllShardNames()){ + public void broadcast(final Function messageSupplier, final Class messageClass) { + for (final String shardName : configuration.getAllShardNames()) { Future primaryFuture = findPrimaryShardAsync(shardName); primaryFuture.onComplete(new OnComplete() { @Override - public void onComplete(Throwable failure, PrimaryShardInfo primaryShardInfo) { - if(failure != null) { - LOG.warn("broadcast failed to send message {} to shard {}: {}", - message.getClass().getSimpleName(), shardName, failure); + public void onComplete(final Throwable failure, final PrimaryShardInfo primaryShardInfo) { + if (failure != null) { + LOG.warn("broadcast failed to send message {} to shard {}", messageClass.getSimpleName(), + shardName, failure); } else { + Object message = messageSupplier.apply(primaryShardInfo.getPrimaryShardVersion()); primaryShardInfo.getPrimaryShardActor().tell(message, ActorRef.noSender()); } } @@ -410,7 +435,11 @@ public class ActorContext { return operationDuration; } - public boolean isPathLocal(String path) { + public Timeout getOperationTimeout() { + return operationTimeout; + } + + public boolean isPathLocal(final String path) { if (Strings.isNullOrEmpty(path)) { return false; } @@ -437,101 +466,68 @@ public class ActorContext { } } - /** - * @deprecated This method is present only to support backward compatibility with Helium and should not be - * used any further - * - * - * @param primaryPath - * @param localPathOfRemoteActor - * @return - */ - @Deprecated - public String resolvePath(final String primaryPath, - final String localPathOfRemoteActor) { - StringBuilder builder = new StringBuilder(); - String[] primaryPathElements = primaryPath.split("/"); - builder.append(primaryPathElements[0]).append("//") - .append(primaryPathElements[1]).append(primaryPathElements[2]); - String[] remotePathElements = localPathOfRemoteActor.split("/"); - for (int i = 3; i < remotePathElements.length; i++) { - builder.append("/").append(remotePathElements[i]); - } - - return builder.toString(); - } - - /** - * Get the maximum number of operations that are to be permitted within a transaction before the transaction - * should begin throttling the operations - * - * Parking reading this configuration here because we need to get to the actor system settings - * - * @return - */ - public int getTransactionOutstandingOperationLimit(){ - return transactionOutstandingOperationLimit; - } - /** * This is a utility method that lets us get a Timer object for any operation. This is a little open-ended to allow * us to create a timer for pretty much anything. * - * @param operationName - * @return + * @param operationName the name of the operation + * @return the Timer instance */ - public Timer getOperationTimer(String operationName){ - return getOperationTimer(datastoreContext.getDataStoreType(), operationName); + public Timer getOperationTimer(final String operationName) { + return getOperationTimer(datastoreContext.getDataStoreName(), operationName); } - public Timer getOperationTimer(String dataStoreType, String operationName){ + public Timer getOperationTimer(final String dataStoreType, final String operationName) { final String rate = MetricRegistry.name(DISTRIBUTED_DATA_STORE_METRIC_REGISTRY, dataStoreType, operationName, METRIC_RATE); return metricRegistry.timer(rate); } /** - * Get the type of the data store to which this ActorContext belongs + * Get the name of the data store to which this ActorContext belongs. * - * @return + * @return the data store name */ - public String getDataStoreType() { - return datastoreContext.getDataStoreType(); + public String getDataStoreName() { + return datastoreContext.getDataStoreName(); } /** - * Get the current transaction creation rate limit - * @return + * Get the current transaction creation rate limit. + * + * @return the rate limit */ - public double getTxCreationLimit(){ + public double getTxCreationLimit() { return txRateLimiter.getTxCreationLimit(); } /** * Try to acquire a transaction creation permit. Will block if no permits are available. */ - public void acquireTxCreationPermit(){ + public void acquireTxCreationPermit() { txRateLimiter.acquire(); } /** - * Return the operation timeout to be used when committing transactions - * @return + * Returns the operation timeout to be used when committing transactions. + * + * @return the operation timeout */ - public Timeout getTransactionCommitOperationTimeout(){ + public Timeout getTransactionCommitOperationTimeout() { return transactionCommitOperationTimeout; } /** * An akka dispatcher that is meant to be used when processing ask Futures which were triggered by client - * code on the datastore - * @return + * code on the datastore. + * + * @return the dispatcher */ public ExecutionContext getClientDispatcher() { return this.dispatchers.getDispatcher(Dispatchers.DispatcherType.Client); } - public String getNotificationDispatcherPath(){ + public String getNotificationDispatcherPath() { return this.dispatchers.getDispatcherPath(Dispatchers.DispatcherType.Notification); } @@ -539,11 +535,15 @@ public class ActorContext { return configuration; } - protected Future doAsk(ActorRef actorRef, Object message, Timeout timeout){ + public ShardStrategyFactory getShardStrategyFactory() { + return shardStrategyFactory; + } + + protected Future doAsk(final ActorRef actorRef, final Object message, final Timeout timeout) { return ask(actorRef, message, timeout); } - protected Future doAsk(ActorSelection actorRef, Object message, Timeout timeout){ + protected Future doAsk(final ActorSelection actorRef, final Object message, final Timeout timeout) { return ask(actorRef, message, timeout); }