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=904dcdf43989bebfeb180b5f56c489cbf035621a;hp=e12a9663d1fca8a969c25166f38986ea19eab51a;hb=f5a373c5378af41f62a2c36ced4046fbdb77e00b;hpb=51e91f6bdcc88c5aa96f956e516d31dbb5e5d5e0 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 e12a9663d1..904dcdf439 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 @@ -13,26 +13,35 @@ import akka.actor.ActorRef; import akka.actor.ActorSelection; import akka.actor.ActorSystem; 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 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; import org.slf4j.LoggerFactory; - 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; /** @@ -45,23 +54,55 @@ public class ActorContext { private static final Logger LOG = LoggerFactory.getLogger(ActorContext.class); - public static final FiniteDuration ASK_DURATION = - Duration.create(5, TimeUnit.SECONDS); - public static final Duration AWAIT_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 final FiniteDuration operationDuration; + private final Timeout operationTimeout; 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; + + operationDuration = Duration.create(datastoreContext.getOperationTimeoutInSeconds(), + TimeUnit.SECONDS); + operationTimeout = new Timeout(operationDuration); + } + + public DatastoreContext getDatastoreContext() { + return datastoreContext; } public ActorSystem getActorSystem() { @@ -80,53 +121,128 @@ public class ActorContext { return actorSystem.actorSelection(actorPath); } + public void setSchemaContext(SchemaContext schemaContext) { + this.schemaContext = schemaContext; + + if(shardManager != null) { + shardManager.tell(new UpdateSchemaContext(schemaContext), null); + } + } + + public SchemaContext getSchemaContext() { + return schemaContext; + } /** - * Finds the primary for a given shard + * Finds the primary shard for the given shard name * * @param shardName * @return */ - public ActorSelection findPrimary(String shardName) { - String path = findPrimaryPath(shardName); - return actorSystem.actorSelection(path); + public Optional findPrimaryShard(String shardName) { + String path = findPrimaryPathOrNull(shardName); + if (path == null){ + return Optional.absent(); + } + 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 ActorRef findLocalShard(String shardName) { - Object result = executeLocalOperation(shardManager, - new FindLocalShard(shardName), ASK_DURATION); + 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(); } + /** + * 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()); + } - public String findPrimaryPath(String shardName) { - Object result = executeLocalOperation(shardManager, - new FindPrimary(shardName).toSerializable(), ASK_DURATION); + private String findPrimaryPathOrNull(String shardName) { + Object result = executeOperation(shardManager, new FindPrimary(shardName, false).toSerializable()); if (result.getClass().equals(PrimaryFound.SERIALIZABLE_CLASS)) { PrimaryFound found = PrimaryFound.fromSerializable(result); LOG.debug("Primary found {}", found.getPrimaryPath()); - return found.getPrimaryPath(); + + } else if (result.getClass().equals(ActorNotInitialized.class)){ + throw new NotInitializedException( + String.format("Found primary shard[%s] but its not initialized yet. Please try again later", shardName) + ); + + } else { + return null; } - throw new PrimaryNotFoundException("Could not find primary for shardName " + shardName); } @@ -135,41 +251,42 @@ public class ActorContext { * * @param actor * @param message - * @param duration * @return The response of the operation */ - public Object executeLocalOperation(ActorRef actor, Object message, - FiniteDuration duration) { - Future future = - ask(actor, message, new Timeout(duration)); + public Object executeOperation(ActorRef actor, Object message) { + Future future = executeOperationAsync(actor, message, operationTimeout); try { - return Await.result(future, AWAIT_DURATION); + 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().toString(), actor.toString()); + return ask(actor, message, timeout); + } + /** * Execute an operation on a remote actor and wait for it's response * * @param actor * @param message - * @param duration * @return */ - public Object executeRemoteOperation(ActorSelection actor, Object message, - FiniteDuration duration) { - - LOG.debug("Sending remote message {} to {}", message.getClass().toString(), actor.toString()); - - Future future = - ask(actor, message, new Timeout(duration)); + public Object executeOperation(ActorSelection actor, Object message) { + Future future = executeOperationAsync(actor, message); try { - return Await.result(future, AWAIT_DURATION); + 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); } } @@ -178,15 +295,28 @@ public class ActorContext { * * @param actor the ActorSelection * @param message the message to send - * @param duration the maximum amount of time to send he message + * @param timeout the operation timeout * @return a Future containing the eventual result */ - public Future executeRemoteOperationAsync(ActorSelection actor, Object message, - FiniteDuration duration) { + 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 remote message {} to {}", message.getClass().toString(), actor.toString()); + LOG.debug("Sending message {} to {}", message.getClass().toString(), actor.toString()); + + return ask(actor, message, timeout); + } - return ask(actor, message, new Timeout(duration)); + /** + * 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); } /** @@ -196,94 +326,97 @@ 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().toString(), actor.toString()); + actor.tell(message, ActorRef.noSender()); } + public void shutdown() { + shardManager.tell(PoisonPill.getInstance(), null); + actorSystem.shutdown(); + } + + 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 - * @param duration - * @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, - FiniteDuration duration) { - ActorSelection primary = findPrimary(shardName); + public void broadcast(Object message){ + for(String shardName : configuration.getAllShardNames()){ + + Optional primary = findPrimaryShard(shardName); + if (primary.isPresent()) { + primary.get().tell(message, ActorRef.noSender()); + } else { + LOG.warn("broadcast failed to send message {} to shard {}. Primary not found", + message.getClass().getSimpleName(), shardName); + } + } + } - return executeRemoteOperation(primary, message, duration); + public FiniteDuration getOperationDuration() { + return operationDuration; } - /** - * 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 - * @param duration the time duration in which this operation should complete - * @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, - FiniteDuration duration) { - ActorRef local = findLocalShard(shardName); + public boolean isLocalPath(String path) { + String selfAddress = clusterWrapper.getSelfAddress(); + if (path == null || selfAddress == null) { + return false; + } + + int atIndex1 = path.indexOf("@"); + int atIndex2 = selfAddress.indexOf("@"); - if(local != null) { - return executeLocalOperation(local, message, duration); + if (atIndex1 == -1 || atIndex2 == -1) { + return false; } - return null; - } + int slashIndex1 = path.indexOf("/", atIndex1); + int slashIndex2 = selfAddress.indexOf("/", atIndex2); + + if (slashIndex1 == -1 || slashIndex2 == -1) { + return false; + } + String hostPort1 = path.substring(atIndex1, slashIndex1); + String hostPort2 = selfAddress.substring(atIndex2, slashIndex2); - public void shutdown() { - shardManager.tell(PoisonPill.getInstance(), null); - actorSystem.shutdown(); + return hostPort1.equals(hostPort2); } /** - * @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(); - - } - - public ActorPath actorFor(String path){ - return actorSystem.actorFor(path).path(); } - - public String getCurrentMemberName(){ - return clusterWrapper.getCurrentMemberName(); - } - }