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%2Futils%2FActorContext.java;h=f1a1f145b370720c80e0a9fd1cb619b44cc4c52a;hp=876901cbf8bbec8f1094f73236af8fc6e33e954f;hb=925cb4a228d0fda99c7bfeb432eb25285a223887;hpb=4e3f49788c05730b29468deebc2aaa4ed0d94eef 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 876901cbf8..f1a1f145b3 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 @@ -9,6 +9,7 @@ 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; @@ -74,11 +75,11 @@ public class ActorContext { @Override public Throwable apply(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; @@ -101,8 +102,13 @@ public class ActorContext { 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; @@ -171,7 +177,7 @@ public class ActorContext { public void setSchemaContext(SchemaContext schemaContext) { this.schemaContext = schemaContext; - if(shardManager != null) { + if (shardManager != null) { shardManager.tell(new UpdateSchemaContext(schemaContext), ActorRef.noSender()); } } @@ -189,7 +195,7 @@ public class ActorContext { updated = true; - if(shardManager != null) { + if (shardManager != null) { shardManager.tell(contextFactory, ActorRef.noSender()); } } @@ -200,7 +206,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, @@ -208,21 +214,21 @@ public class ActorContext { return future.transform(new Mapper() { @Override - public PrimaryShardInfo checkedApply(Object response) throws Exception { - if(response instanceof RemotePrimaryShardFound) { + public PrimaryShardInfo checkedApply(Object response) throws UnknownMessageException { + if (response instanceof RemotePrimaryShardFound) { LOG.debug("findPrimaryShardAsync received: {}", response); RemotePrimaryShardFound found = (RemotePrimaryShardFound)response; return onPrimaryShardFound(shardName, found.getPrimaryPath(), found.getPrimaryVersion(), null); - } else if(response instanceof LocalPrimaryShardFound) { + } else if (response instanceof LocalPrimaryShardFound) { LOG.debug("findPrimaryShardAsync received: {}", response); LocalPrimaryShardFound found = (LocalPrimaryShardFound)response; return onPrimaryShardFound(shardName, found.getPrimaryPath(), DataStoreVersions.CURRENT_VERSION, found.getLocalShardDataTree()); - } else if(response instanceof NotInitializedException) { + } 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; } @@ -242,7 +248,7 @@ public class ActorContext { } /** - * 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 @@ -273,13 +279,13 @@ public class ActorContext { return future.map(new Mapper() { @Override public ActorRef checkedApply(Object response) throws Throwable { - if(response instanceof LocalShardFound) { + 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)); } @@ -291,49 +297,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 */ + @SuppressWarnings("checkstyle:IllegalCatch") 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. 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 */ + @SuppressWarnings("checkstyle:IllegalCatch") 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. 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 asynchronously. * @@ -379,11 +387,12 @@ public class ActorContext { actor.tell(message, ActorRef.noSender()); } + @SuppressWarnings("checkstyle:IllegalCatch") public void shutdown() { FiniteDuration duration = datastoreContext.getShardRaftConfig().getElectionTimeOutInterval().$times(3); try { Await.ready(Patterns.gracefulStop(shardManager, duration, Shutdown.INSTANCE), duration); - } catch(Exception e) { + } catch (Exception e) { LOG.warn("ShardManager for {} data store did not shutdown gracefully", getDataStoreName(), e); } } @@ -392,27 +401,25 @@ public class ActorContext { return clusterWrapper; } - public MemberName 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 Function messageSupplier){ - for(final String shardName : configuration.getAllShardNames()){ + public void broadcast(final Function messageSupplier, Class messageClass) { + for (final String shardName : configuration.getAllShardNames()) { Future primaryFuture = findPrimaryShardAsync(shardName); primaryFuture.onComplete(new OnComplete() { @Override public void onComplete(Throwable failure, PrimaryShardInfo primaryShardInfo) { - Object message = messageSupplier.apply(primaryShardInfo.getPrimaryShardVersion()); - if(failure != null) { + if (failure != null) { LOG.warn("broadcast failed to send message {} to shard {}: {}", - message.getClass().getSimpleName(), shardName, failure); + messageClass.getSimpleName(), shardName, failure); } else { + Object message = messageSupplier.apply(primaryShardInfo.getPrimaryShardVersion()); primaryShardInfo.getPrimaryShardActor().tell(message, ActorRef.noSender()); } } @@ -459,61 +466,64 @@ public class ActorContext { * 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){ + public Timer getOperationTimer(String operationName) { return getOperationTimer(datastoreContext.getDataStoreName(), operationName); } - public Timer getOperationTimer(String dataStoreType, String operationName){ + public Timer getOperationTimer(String dataStoreType, String operationName) { final String rate = MetricRegistry.name(DISTRIBUTED_DATA_STORE_METRIC_REGISTRY, dataStoreType, operationName, METRIC_RATE); return metricRegistry.timer(rate); } /** - * Get the name 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 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); } @@ -525,11 +535,11 @@ public class ActorContext { return shardStrategyFactory; } - protected Future doAsk(ActorRef actorRef, Object message, Timeout timeout){ + protected Future doAsk(ActorRef actorRef, Object message, Timeout timeout) { return ask(actorRef, message, timeout); } - protected Future doAsk(ActorSelection actorRef, Object message, Timeout timeout){ + protected Future doAsk(ActorSelection actorRef, Object message, Timeout timeout) { return ask(actorRef, message, timeout); }