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=44f4ef77d7ff057dbea32ebfd9d3404f5b30f862;hb=c9d0ae2a8ce525d150f360cf49f21b7c0187cfbf;hp=7b5588cb196a66fa68de947fecc58137977275ae;hpb=4a8688c9d42c0be307383f0483c819cb7a78d26d;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 7b5588cb19..b6250fc1cc 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 @@ -8,34 +8,54 @@ 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.pattern.Patterns; +import akka.dispatch.Futures; +import akka.dispatch.Mapper; +import akka.dispatch.OnComplete; +import akka.pattern.AskTimeoutException; import akka.util.Timeout; +import com.codahale.metrics.JmxReporter; +import com.codahale.metrics.MetricRegistry; +import com.codahale.metrics.Timer; +import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Optional; +import com.google.common.base.Preconditions; +import com.google.common.base.Strings; +import com.google.common.cache.Cache; +import com.google.common.cache.CacheBuilder; +import com.google.common.util.concurrent.RateLimiter; +import java.util.concurrent.TimeUnit; +import org.opendaylight.controller.cluster.common.actor.CommonConfig; import org.opendaylight.controller.cluster.datastore.ClusterWrapper; import org.opendaylight.controller.cluster.datastore.Configuration; +import org.opendaylight.controller.cluster.datastore.DatastoreContext; +import org.opendaylight.controller.cluster.datastore.exceptions.LocalShardNotFoundException; +import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException; +import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedException; import org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException; import org.opendaylight.controller.cluster.datastore.exceptions.TimeoutException; +import org.opendaylight.controller.cluster.datastore.exceptions.UnknownMessageException; import org.opendaylight.controller.cluster.datastore.messages.FindLocalShard; import org.opendaylight.controller.cluster.datastore.messages.FindPrimary; import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound; +import org.opendaylight.controller.cluster.datastore.messages.LocalShardNotFound; import org.opendaylight.controller.cluster.datastore.messages.PrimaryFound; import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext; import org.opendaylight.yangtools.yang.model.api.SchemaContext; import org.slf4j.Logger; 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; -import java.util.concurrent.TimeUnit; - -import static akka.pattern.Patterns.ask; - /** * The ActorContext class contains utility methods which could be used by * non-actors (like DistributedDataStore) to work with actors a little more @@ -43,28 +63,96 @@ import static akka.pattern.Patterns.ask; * but should not be passed to actors especially remote actors */ public class ActorContext { - private static final Logger - LOG = LoggerFactory.getLogger(ActorContext.class); - - private static final FiniteDuration DEFAULT_OPER_DURATION = Duration.create(5, TimeUnit.SECONDS); + private static final Logger LOG = LoggerFactory.getLogger(ActorContext.class); + private static final String UNKNOWN_DATA_STORE_TYPE = "unknown"; + private static final String DISTRIBUTED_DATA_STORE_METRIC_REGISTRY = "distributed-data-store"; + private static final String METRIC_RATE = "rate"; + private static final String DOMAIN = "org.opendaylight.controller.cluster.datastore"; + private static final Mapper FIND_PRIMARY_FAILURE_TRANSFORMER = + new Mapper() { + @Override + public Throwable apply(Throwable failure) { + Throwable actualFailure = failure; + 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."); + } + return actualFailure; + } + }; public static final String MAILBOX = "bounded-mailbox"; private final ActorSystem actorSystem; private final ActorRef shardManager; private final ClusterWrapper clusterWrapper; private final Configuration configuration; + private DatastoreContext datastoreContext; + private FiniteDuration operationDuration; + private Timeout operationTimeout; + private final String selfAddressHostPort; + private RateLimiter txRateLimiter; + private final MetricRegistry metricRegistry = new MetricRegistry(); + private final JmxReporter jmxReporter = JmxReporter.forRegistry(metricRegistry).inDomain(DOMAIN).build(); + private final int transactionOutstandingOperationLimit; + private Timeout transactionCommitOperationTimeout; + private Timeout shardInitializationTimeout; + private final Dispatchers dispatchers; + private Cache> primaryShardActorSelectionCache; + private volatile SchemaContext schemaContext; - private FiniteDuration operationDuration = DEFAULT_OPER_DURATION; - private Timeout operationTimeout = new Timeout(operationDuration); + private volatile boolean updated; public ActorContext(ActorSystem actorSystem, ActorRef shardManager, - ClusterWrapper clusterWrapper, - Configuration configuration) { + ClusterWrapper clusterWrapper, Configuration configuration) { + this(actorSystem, shardManager, clusterWrapper, configuration, + DatastoreContext.newBuilder().build()); + } + + public ActorContext(ActorSystem actorSystem, ActorRef shardManager, + ClusterWrapper clusterWrapper, Configuration configuration, + DatastoreContext datastoreContext) { this.actorSystem = actorSystem; this.shardManager = shardManager; this.clusterWrapper = clusterWrapper; this.configuration = configuration; + this.datastoreContext = datastoreContext; + this.dispatchers = new Dispatchers(actorSystem.dispatchers()); + + setCachedProperties(); + + Address selfAddress = clusterWrapper.getSelfAddress(); + if (selfAddress != null && !selfAddress.host().isEmpty()) { + selfAddressHostPort = selfAddress.host().get() + ":" + selfAddress.port().get(); + } else { + selfAddressHostPort = null; + } + + transactionOutstandingOperationLimit = new CommonConfig(this.getActorSystem().settings().config()).getMailBoxCapacity(); + jmxReporter.start(); + + } + + private void setCachedProperties() { + txRateLimiter = RateLimiter.create(datastoreContext.getTransactionCreationInitialRateLimit()); + + operationDuration = Duration.create(datastoreContext.getOperationTimeoutInSeconds(), TimeUnit.SECONDS); + operationTimeout = new Timeout(operationDuration); + + transactionCommitOperationTimeout = new Timeout(Duration.create( + datastoreContext.getShardTransactionCommitTimeoutInSeconds(), TimeUnit.SECONDS)); + + shardInitializationTimeout = new Timeout(datastoreContext.getShardInitializationTimeout().duration().$times(2)); + + primaryShardActorSelectionCache = CacheBuilder.newBuilder() + .expireAfterWrite(datastoreContext.getShardLeaderElectionTimeout().duration().toMillis(), TimeUnit.MILLISECONDS) + .build(); + } + + public DatastoreContext getDatastoreContext() { + return datastoreContext; } public ActorSystem getActorSystem() { @@ -87,68 +175,113 @@ public class ActorContext { this.schemaContext = schemaContext; if(shardManager != null) { - shardManager.tell(new UpdateSchemaContext(schemaContext), null); + shardManager.tell(new UpdateSchemaContext(schemaContext), ActorRef.noSender()); } } - public void setOperationTimeout(int timeoutInSeconds) { - operationDuration = Duration.create(timeoutInSeconds, TimeUnit.SECONDS); - operationTimeout = new Timeout(operationDuration); + public void setDatastoreContext(DatastoreContext context) { + this.datastoreContext = context; + setCachedProperties(); + + // We write the 'updated' volatile to trigger a write memory barrier so that the writes above + // will be published immediately even though they may not be immediately visible to other + // threads due to unsynchronized reads. That's OK though - we're going for eventual + // consistency here as immediately visible updates to these members aren't critical. These + // members could've been made volatile but wanted to avoid volatile reads as these are + // accessed often and updates will be infrequent. + + updated = true; + + if(shardManager != null) { + shardManager.tell(context, ActorRef.noSender()); + } } public SchemaContext getSchemaContext() { return schemaContext; } - /** - * Finds the primary for a given shard - * - * @param shardName - * @return - */ - public ActorSelection findPrimary(String shardName) { - String path = findPrimaryPath(shardName); - return actorSystem.actorSelection(path); + public Future findPrimaryShardAsync(final String shardName) { + Future ret = primaryShardActorSelectionCache.getIfPresent(shardName); + if(ret != null){ + return ret; + } + Future future = executeOperationAsync(shardManager, + new FindPrimary(shardName, true), shardInitializationTimeout); + + return future.transform(new Mapper() { + @Override + public ActorSelection checkedApply(Object response) throws Exception { + if(response instanceof PrimaryFound) { + PrimaryFound found = (PrimaryFound)response; + + LOG.debug("Primary found {}", found.getPrimaryPath()); + ActorSelection actorSelection = actorSystem.actorSelection(found.getPrimaryPath()); + primaryShardActorSelectionCache.put(shardName, Futures.successful(actorSelection)); + return actorSelection; + } else if(response instanceof NotInitializedException) { + throw (NotInitializedException)response; + } else if(response instanceof PrimaryNotFoundException) { + throw (PrimaryNotFoundException)response; + } else if(response instanceof NoShardLeaderException) { + throw (NoShardLeaderException)response; + } + + throw new UnknownMessageException(String.format( + "FindPrimary returned unkown response: %s", response)); + } + }, FIND_PRIMARY_FAILURE_TRANSFORMER, getClientDispatcher()); } /** - * Finds a local shard given it's 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 ActorRef findLocalShard(String shardName) { - Object result = executeLocalOperation(shardManager, - new FindLocalShard(shardName)); + public Optional findLocalShard(String shardName) { + Object result = executeOperation(shardManager, new FindLocalShard(shardName, false)); if (result instanceof LocalShardFound) { LocalShardFound found = (LocalShardFound) result; - LOG.debug("Local shard found {}", found.getPath()); - - return found.getPath(); + return Optional.of(found.getPath()); } - return null; + return Optional.absent(); } - - public String findPrimaryPath(String shardName) { - Object result = executeLocalOperation(shardManager, - new FindPrimary(shardName).toSerializable()); - - if (result.getClass().equals(PrimaryFound.SERIALIZABLE_CLASS)) { - PrimaryFound found = PrimaryFound.fromSerializable(result); - - LOG.debug("Primary found {}", found.getPrimaryPath()); - - return found.getPrimaryPath(); - } - throw new PrimaryNotFoundException("Could not find primary for shardName " + shardName); + /** + * Finds a local shard async given its shard name and return a Future from which to obtain the + * ActorRef. + * + * @param shardName the name of the local shard that needs to be found + */ + 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) { + LocalShardFound found = (LocalShardFound)response; + LOG.debug("Local shard found {}", found.getPath()); + return found.getPath(); + } else if(response instanceof NotInitializedException) { + throw (NotInitializedException)response; + } else if(response instanceof LocalShardNotFound) { + throw new LocalShardNotFoundException( + String.format("Local shard for %s does not exist.", shardName)); + } + + throw new UnknownMessageException(String.format( + "FindLocalShard returned unkown response: %s", response)); + } + }, getClientDispatcher()); } - /** * Executes an operation on a local actor and wait for it's response * @@ -156,16 +289,25 @@ public class ActorContext { * @param message * @return The response of the operation */ - public Object executeLocalOperation(ActorRef actor, Object message) { - Future future = ask(actor, message, operationTimeout); + public Object executeOperation(ActorRef actor, 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" , 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 * @@ -173,18 +315,14 @@ public class ActorContext { * @param message * @return */ - public Object executeRemoteOperation(ActorSelection actor, Object message) { - - LOG.debug("Sending remote message {} to {}", message.getClass().toString(), - actor.toString()); - - Future future = ask(actor, message, operationTimeout); + public Object executeOperation(ActorSelection actor, 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" , e); + " to actor " + actor.toString() + " failed. Try again later.", e); } } @@ -193,13 +331,28 @@ public class ActorContext { * * @param actor the ActorSelection * @param message the message to send + * @param timeout the operation timeout * @return a Future containing the eventual result */ - public Future executeRemoteOperationAsync(ActorSelection actor, Object message) { + public Future executeOperationAsync(ActorSelection 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); - LOG.debug("Sending remote message {} to {}", message.getClass().toString(), actor.toString()); + return doAsk(actor, message, timeout); + } - return ask(actor, message, operationTimeout); + /** + * Execute an operation on a remote actor asynchronously. + * + * @param actor the ActorSelection + * @param message the message to send + * @return a Future containing the eventual result + */ + public Future executeOperationAsync(ActorSelection actor, Object message) { + return executeOperationAsync(actor, message, operationTimeout); } /** @@ -209,139 +362,194 @@ public class ActorContext { * @param actor the ActorSelection * @param message the message to send */ - public void sendRemoteOperationAsync(ActorSelection actor, Object message) { + public void sendOperationAsync(ActorSelection actor, Object message) { + 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); + actor.tell(message, ActorRef.noSender()); } - public void sendShardOperationAsync(String shardName, Object message) { - ActorSelection primary = findPrimary(shardName); + public void shutdown() { + shardManager.tell(PoisonPill.getInstance(), null); + actorSystem.shutdown(); + } - primary.tell(message, ActorRef.noSender()); + public ClusterWrapper getClusterWrapper() { + return clusterWrapper; } + public String getCurrentMemberName(){ + return clusterWrapper.getCurrentMemberName(); + } /** - * Execute an operation on the primary for a given shard - *

- * This method first finds the primary for a given shard ,then sends - * the message to the remote shard and waits for a response - *

+ * Send the message to each and every shard * - * @param shardName * @param message - * @return - * @throws org.opendaylight.controller.cluster.datastore.exceptions.TimeoutException if the message to the remote shard times out - * @throws org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException if the primary shard is not found */ - public Object executeShardOperation(String shardName, Object message) { - ActorSelection primary = findPrimary(shardName); - - return executeRemoteOperation(primary, message); + public void broadcast(final Object message){ + for(final String shardName : configuration.getAllShardNames()){ + + Future primaryFuture = findPrimaryShardAsync(shardName); + primaryFuture.onComplete(new OnComplete() { + @Override + public void onComplete(Throwable failure, ActorSelection primaryShard) { + if(failure != null) { + LOG.warn("broadcast failed to send message {} to shard {}: {}", + message.getClass().getSimpleName(), shardName, failure); + } else { + primaryShard.tell(message, ActorRef.noSender()); + } + } + }, getClientDispatcher()); + } } - /** - * Execute an operation on the the local shard only - *

- * This method first finds the address of the local shard if any. It then - * executes the operation on it. - *

- * - * @param shardName the name of the shard on which the operation needs to be executed - * @param message the message that needs to be sent to the shard - * @return the message that was returned by the local actor on which the - * the operation was executed. If a local shard was not found then - * null is returned - * @throws org.opendaylight.controller.cluster.datastore.exceptions.TimeoutException - * if the operation does not complete in a specified time duration - */ - public Object executeLocalShardOperation(String shardName, Object message) { - ActorRef local = findLocalShard(shardName); + public FiniteDuration getOperationDuration() { + return operationDuration; + } - if(local != null) { - return executeLocalOperation(local, message); + public boolean isPathLocal(String path) { + if (Strings.isNullOrEmpty(path)) { + return false; } - return null; - } - + int pathAtIndex = path.indexOf('@'); + if (pathAtIndex == -1) { + //if the path is of local format, then its local and is co-located + return true; - /** - * Execute an operation on the the local shard only asynchronously - * - *

- * This method first finds the address of the local shard if any. It then - * executes the operation on it. - *

- * - * @param shardName the name of the shard on which the operation needs to be executed - * @param message the message that needs to be sent to the shard - * @param timeout the amount of time that this method should wait for a response before timing out - * @return null if the shard could not be located else a future on which the caller can wait - * - */ - public Future executeLocalShardOperationAsync(String shardName, Object message, Timeout timeout) { - ActorRef local = findLocalShard(shardName); - if(local == null){ - return null; - } - return Patterns.ask(local, message, timeout); - } + } else if (selfAddressHostPort != null) { + // self-address and tx actor path, both are of remote path format + int slashIndex = path.indexOf('/', pathAtIndex); + if (slashIndex == -1) { + return false; + } + String hostPort = path.substring(pathAtIndex + 1, slashIndex); + return hostPort.equals(selfAddressHostPort); - public void shutdown() { - shardManager.tell(PoisonPill.getInstance(), null); - actorSystem.shutdown(); + } else { + // self address is local format and tx actor path is remote format + return false; + } } /** - * @deprecated Need to stop using this method. There are ways to send a - * remote ActorRef as a string which should be used instead of this hack + * @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) { + 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]); - } + 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; } - public ActorPath actorFor(String path){ - return actorSystem.actorFor(path).path(); + /** + * 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 + */ + public Timer getOperationTimer(String operationName){ + final String rate = MetricRegistry.name(DISTRIBUTED_DATA_STORE_METRIC_REGISTRY, datastoreContext.getDataStoreType(), operationName, METRIC_RATE); + return metricRegistry.timer(rate); } - public String getCurrentMemberName(){ - return clusterWrapper.getCurrentMemberName(); + /** + * Get the type of the data store to which this ActorContext belongs + * + * @return + */ + public String getDataStoreType() { + return datastoreContext.getDataStoreType(); } /** - * Send the message to each and every shard + * Set the number of transaction creation permits that are to be allowed * - * @param message + * @param permitsPerSecond */ - public void broadcast(Object message){ - for(String shardName : configuration.getAllShardNames()){ - try { - sendShardOperationAsync(shardName, message); - } catch(Exception e){ - LOG.warn("broadcast failed to send message " + message.getClass().getSimpleName() + " to shard " + shardName, e); - } - } + public void setTxCreationLimit(double permitsPerSecond){ + txRateLimiter.setRate(permitsPerSecond); } - public FiniteDuration getOperationDuration() { - return operationDuration; + /** + * Get the current transaction creation rate limit + * @return + */ + public double getTxCreationLimit(){ + return txRateLimiter.getRate(); + } + + /** + * Try to acquire a transaction creation permit. Will block if no permits are available. + */ + public void acquireTxCreationPermit(){ + txRateLimiter.acquire(); + } + + /** + * Return the operation timeout to be used when committing transactions + * @return + */ + 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 + */ + public ExecutionContext getClientDispatcher() { + return this.dispatchers.getDispatcher(Dispatchers.DispatcherType.Client); + } + + public String getNotificationDispatcherPath(){ + return this.dispatchers.getDispatcherPath(Dispatchers.DispatcherType.Notification); + } + + protected Future doAsk(ActorRef actorRef, Object message, Timeout timeout){ + return ask(actorRef, message, timeout); + } + + protected Future doAsk(ActorSelection actorRef, Object message, Timeout timeout){ + return ask(actorRef, message, timeout); + } + + @VisibleForTesting + Cache> getPrimaryShardActorSelectionCache() { + return primaryShardActorSelectionCache; } }