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=f217d05bb21a12e6f92add47da5536c6f6fe12d9;hb=fcf65d723ef53f8da2dd6347f41ce19016fc36e5;hp=314ae916de1444349816988089011d82b7cd9c8b;hpb=26ead80b71a367f8dac348905d440d7e203e0b4f;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 314ae916de..f217d05bb2 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 @@ -12,19 +12,30 @@ 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.pattern.AskTimeoutException; import akka.util.Timeout; 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.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.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.ActorNotInitialized; 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.PrimaryNotFound; import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext; import org.opendaylight.yangtools.yang.model.api.SchemaContext; import org.slf4j.Logger; @@ -33,7 +44,7 @@ import scala.concurrent.Await; 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; /** @@ -46,25 +57,63 @@ public class ActorContext { private static final Logger LOG = LoggerFactory.getLogger(ActorContext.class); - private static final FiniteDuration DEFAULT_OPER_DURATION = Duration.create(5, TimeUnit.SECONDS); - public static final String MAILBOX = "bounded-mailbox"; + 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; + } + }; + private final ActorSystem actorSystem; private final ActorRef shardManager; private final ClusterWrapper clusterWrapper; private final Configuration configuration; + private final DatastoreContext datastoreContext; private volatile SchemaContext schemaContext; - private FiniteDuration operationDuration = DEFAULT_OPER_DURATION; - private Timeout operationTimeout = new Timeout(operationDuration); + private final FiniteDuration operationDuration; + private final Timeout operationTimeout; + private final String selfAddressHostPort; + + public ActorContext(ActorSystem actorSystem, ActorRef shardManager, + ClusterWrapper clusterWrapper, Configuration configuration) { + this(actorSystem, shardManager, clusterWrapper, configuration, + DatastoreContext.newBuilder().build()); + } public ActorContext(ActorSystem actorSystem, ActorRef shardManager, - ClusterWrapper clusterWrapper, - Configuration configuration) { + ClusterWrapper clusterWrapper, Configuration configuration, + DatastoreContext datastoreContext) { this.actorSystem = actorSystem; this.shardManager = shardManager; this.clusterWrapper = clusterWrapper; this.configuration = configuration; + this.datastoreContext = datastoreContext; + + operationDuration = Duration.create(datastoreContext.getOperationTimeoutInSeconds(), + TimeUnit.SECONDS); + operationTimeout = new Timeout(operationDuration); + + Address selfAddress = clusterWrapper.getSelfAddress(); + if (selfAddress != null && !selfAddress.host().isEmpty()) { + selfAddressHostPort = selfAddress.host().get() + ":" + selfAddress.port().get(); + } else { + selfAddressHostPort = null; + } + } + + public DatastoreContext getDatastoreContext() { + return datastoreContext; } public ActorSystem getActorSystem() { @@ -91,11 +140,6 @@ public class ActorContext { } } - public void setOperationTimeout(int timeoutInSeconds) { - operationDuration = Duration.create(timeoutInSeconds, TimeUnit.SECONDS); - operationTimeout = new Timeout(operationDuration); - } - public SchemaContext getSchemaContext() { return schemaContext; } @@ -114,15 +158,43 @@ public class ActorContext { return Optional.of(actorSystem.actorSelection(path)); } + public Future findPrimaryShardAsync(final String shardName) { + Future future = executeOperationAsync(shardManager, + new FindPrimary(shardName, true).toSerializable(), + datastoreContext.getShardInitializationTimeout()); + + return future.transform(new Mapper() { + @Override + public ActorSelection checkedApply(Object response) throws Exception { + if(response.getClass().equals(PrimaryFound.SERIALIZABLE_CLASS)) { + PrimaryFound found = PrimaryFound.fromSerializable(response); + + LOG.debug("Primary found {}", found.getPrimaryPath()); + return actorSystem.actorSelection(found.getPrimaryPath()); + } else if(response instanceof ActorNotInitialized) { + throw new NotInitializedException( + String.format("Found primary shard %s but it's not initialized yet. " + + "Please try again later", shardName)); + } else if(response instanceof PrimaryNotFound) { + throw new PrimaryNotFoundException( + String.format("No primary shard found for %S.", shardName)); + } + + throw new UnknownMessageException(String.format( + "FindPrimary returned unkown response: %s", response)); + } + }, FIND_PRIMARY_FAILURE_TRANSFORMER, getActorSystem().dispatcher()); + } + /** - * 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 Optional findLocalShard(String shardName) { - Object result = executeOperation(shardManager, new FindLocalShard(shardName)); + Object result = executeOperation(shardManager, new FindLocalShard(shardName, false)); if (result instanceof LocalShardFound) { LocalShardFound found = (LocalShardFound) result; @@ -133,9 +205,40 @@ public class ActorContext { return Optional.absent(); } + /** + * 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), datastoreContext.getShardInitializationTimeout()); + + 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 ActorNotInitialized) { + throw new NotInitializedException( + String.format("Found local shard for %s but it's not initialized yet.", + shardName)); + } 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)); + } + }, getActorSystem().dispatcher()); + } private String findPrimaryPathOrNull(String shardName) { - Object result = executeOperation(shardManager, new FindPrimary(shardName).toSerializable()); + Object result = executeOperation(shardManager, new FindPrimary(shardName, false).toSerializable()); if (result.getClass().equals(PrimaryFound.SERIALIZABLE_CLASS)) { PrimaryFound found = PrimaryFound.fromSerializable(result); @@ -176,7 +279,7 @@ public class ActorContext { 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().toString(), actor.toString()); + LOG.debug("Sending message {} to {}", message.getClass(), actor); return ask(actor, message, timeout); } @@ -203,15 +306,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 executeOperationAsync(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().toString(), actor.toString()); + LOG.debug("Sending message {} to {}", message.getClass(), actor); + + return ask(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); } /** @@ -225,7 +341,7 @@ public class ActorContext { 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().toString(), actor.toString()); + LOG.debug("Sending message {} to {}", message.getClass(), actor); actor.tell(message, ActorRef.noSender()); } @@ -265,29 +381,54 @@ public class ActorContext { return operationDuration; } - public boolean isLocalPath(String path) { - String selfAddress = clusterWrapper.getSelfAddress(); - if (path == null || selfAddress == null) { + public boolean isPathLocal(String path) { + if (Strings.isNullOrEmpty(path)) { return false; } - int atIndex1 = path.indexOf("@"); - int atIndex2 = selfAddress.indexOf("@"); + int pathAtIndex = path.indexOf('@'); + if (pathAtIndex == -1) { + //if the path is of local format, then its local and is co-located + return true; - if (atIndex1 == -1 || atIndex2 == -1) { - return false; - } + } else if (selfAddressHostPort != null) { + // self-address and tx actor path, both are of remote path format + int slashIndex = path.indexOf('/', pathAtIndex); - int slashIndex1 = path.indexOf("/", atIndex1); - int slashIndex2 = selfAddress.indexOf("/", atIndex2); + if (slashIndex == -1) { + return false; + } + + String hostPort = path.substring(pathAtIndex + 1, slashIndex); + return hostPort.equals(selfAddressHostPort); - if (slashIndex1 == -1 || slashIndex2 == -1) { + } else { + // self address is local format and tx actor path is remote format return false; } + } - String hostPort1 = path.substring(atIndex1, slashIndex1); - String hostPort2 = selfAddress.substring(atIndex2, slashIndex2); + /** + * @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 hostPort1.equals(hostPort2); + return builder.toString(); } }