import akka.actor.ActorSystem;
import akka.actor.Address;
import akka.actor.PoisonPill;
+import akka.dispatch.Futures;
import akka.dispatch.Mapper;
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;
private final ActorRef shardManager;
private final ClusterWrapper clusterWrapper;
private final Configuration configuration;
- private final DatastoreContext datastoreContext;
- private final FiniteDuration operationDuration;
- private final Timeout operationTimeout;
+ private DatastoreContext datastoreContext;
+ private FiniteDuration operationDuration;
+ private Timeout operationTimeout;
private final String selfAddressHostPort;
- private final RateLimiter txRateLimiter;
+ private RateLimiter txRateLimiter;
private final MetricRegistry metricRegistry = new MetricRegistry();
private final JmxReporter jmxReporter = JmxReporter.forRegistry(metricRegistry).inDomain(DOMAIN).build();
private final int transactionOutstandingOperationLimit;
- private final Timeout transactionCommitOperationTimeout;
+ private Timeout transactionCommitOperationTimeout;
private final Dispatchers dispatchers;
+ private final Cache<String, Future<ActorSelection>> primaryShardActorSelectionCache;
private volatile SchemaContext schemaContext;
+ private volatile boolean updated;
public ActorContext(ActorSystem actorSystem, ActorRef shardManager,
ClusterWrapper clusterWrapper, Configuration configuration) {
this.clusterWrapper = clusterWrapper;
this.configuration = configuration;
this.datastoreContext = datastoreContext;
- this.txRateLimiter = RateLimiter.create(datastoreContext.getTransactionCreationInitialRateLimit());
this.dispatchers = new Dispatchers(actorSystem.dispatchers());
+ setCachedProperties();
+ primaryShardActorSelectionCache = CacheBuilder.newBuilder()
+ .expireAfterWrite(datastoreContext.getShardLeaderElectionTimeout().duration().toMillis(), TimeUnit.MILLISECONDS)
+ .build();
+
operationDuration = Duration.create(datastoreContext.getOperationTimeoutInSeconds(), TimeUnit.SECONDS);
operationTimeout = new Timeout(operationDuration);
transactionCommitOperationTimeout = new Timeout(Duration.create(getDatastoreContext().getShardTransactionCommitTimeoutInSeconds(),
TimeUnit.SECONDS));
-
Address selfAddress = clusterWrapper.getSelfAddress();
if (selfAddress != null && !selfAddress.host().isEmpty()) {
selfAddressHostPort = selfAddress.host().get() + ":" + selfAddress.port().get();
}
+ 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));
+ }
+
public DatastoreContext getDatastoreContext() {
return datastoreContext;
}
this.schemaContext = schemaContext;
if(shardManager != null) {
- shardManager.tell(new UpdateSchemaContext(schemaContext), null);
+ shardManager.tell(new UpdateSchemaContext(schemaContext), ActorRef.noSender());
+ }
+ }
+
+ 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 Future<ActorSelection> findPrimaryShardAsync(final String shardName) {
+ Future<ActorSelection> ret = primaryShardActorSelectionCache.getIfPresent(shardName);
+ if(ret != null){
+ return ret;
+ }
Future<Object> future = executeOperationAsync(shardManager,
new FindPrimary(shardName, true).toSerializable(),
datastoreContext.getShardInitializationTimeout());
return future.transform(new Mapper<Object, ActorSelection>() {
@Override
public ActorSelection checkedApply(Object response) throws Exception {
- if(response.getClass().equals(PrimaryFound.SERIALIZABLE_CLASS)) {
+ if(PrimaryFound.SERIALIZABLE_CLASS.isInstance(response)) {
PrimaryFound found = PrimaryFound.fromSerializable(response);
LOG.debug("Primary found {}", found.getPrimaryPath());
- return actorSystem.actorSelection(found.getPrimaryPath());
+ ActorSelection actorSelection = actorSystem.actorSelection(found.getPrimaryPath());
+ primaryShardActorSelectionCache.put(shardName, Futures.successful(actorSelection));
+ return actorSelection;
} else if(response instanceof ActorNotInitialized) {
throw new NotInitializedException(
String.format("Found primary shard %s but it's not initialized yet. " +
Preconditions.checkArgument(message != null, "message must not be null");
LOG.debug("Sending message {} to {}", message.getClass(), actor);
- return ask(actor, message, timeout);
+ return doAsk(actor, message, timeout);
}
/**
LOG.debug("Sending message {} to {}", message.getClass(), actor);
- return ask(actor, message, timeout);
+ return doAsk(actor, message, timeout);
}
/**
return this.dispatchers.getDispatcherPath(Dispatchers.DispatcherType.Notification);
}
+ protected Future<Object> doAsk(ActorRef actorRef, Object message, Timeout timeout){
+ return ask(actorRef, message, timeout);
+ }
+
+ protected Future<Object> doAsk(ActorSelection actorRef, Object message, Timeout timeout){
+ return ask(actorRef, message, timeout);
+ }
+
+ @VisibleForTesting
+ Cache<String, Future<ActorSelection>> getPrimaryShardActorSelectionCache() {
+ return primaryShardActorSelectionCache;
+ }
}
Code Review / controller.git / blobdiff