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=7b5588cb196a66fa68de947fecc58137977275ae;hp=f76430f5a1cb1cdb9e84f3b92f85c923a7111967;hb=9f61e98b036119694dfef0759a7cafc56aae6e86;hpb=945a63ae76181f7cda4b052fa8fcdf115da4cf2d 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 f76430f5a1..7b5588cb19 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,8 +13,8 @@ import akka.actor.ActorRef; import akka.actor.ActorSelection; import akka.actor.ActorSystem; import akka.actor.PoisonPill; +import akka.pattern.Patterns; import akka.util.Timeout; - import org.opendaylight.controller.cluster.datastore.ClusterWrapper; import org.opendaylight.controller.cluster.datastore.Configuration; import org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException; @@ -27,7 +27,6 @@ import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContex 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; @@ -47,16 +46,17 @@ 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); + private static final FiniteDuration DEFAULT_OPER_DURATION = Duration.create(5, TimeUnit.SECONDS); + + public static final String MAILBOX = "bounded-mailbox"; private final ActorSystem actorSystem; private final ActorRef shardManager; private final ClusterWrapper clusterWrapper; private final Configuration configuration; private volatile SchemaContext schemaContext; + private FiniteDuration operationDuration = DEFAULT_OPER_DURATION; + private Timeout operationTimeout = new Timeout(operationDuration); public ActorContext(ActorSystem actorSystem, ActorRef shardManager, ClusterWrapper clusterWrapper, @@ -91,6 +91,11 @@ public class ActorContext { } } + public void setOperationTimeout(int timeoutInSeconds) { + operationDuration = Duration.create(timeoutInSeconds, TimeUnit.SECONDS); + operationTimeout = new Timeout(operationDuration); + } + public SchemaContext getSchemaContext() { return schemaContext; } @@ -115,7 +120,7 @@ public class ActorContext { */ public ActorRef findLocalShard(String shardName) { Object result = executeLocalOperation(shardManager, - new FindLocalShard(shardName), ASK_DURATION); + new FindLocalShard(shardName)); if (result instanceof LocalShardFound) { LocalShardFound found = (LocalShardFound) result; @@ -131,7 +136,7 @@ public class ActorContext { public String findPrimaryPath(String shardName) { Object result = executeLocalOperation(shardManager, - new FindPrimary(shardName).toSerializable(), ASK_DURATION); + new FindPrimary(shardName).toSerializable()); if (result.getClass().equals(PrimaryFound.SERIALIZABLE_CLASS)) { PrimaryFound found = PrimaryFound.fromSerializable(result); @@ -149,16 +154,13 @@ 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 executeLocalOperation(ActorRef actor, Object message) { + Future future = ask(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); } @@ -169,21 +171,20 @@ public class ActorContext { * * @param actor * @param message - * @param duration * @return */ - public Object executeRemoteOperation(ActorSelection actor, Object message, - FiniteDuration duration) { + public Object executeRemoteOperation(ActorSelection actor, Object message) { - LOG.debug("Sending remote message {} to {}", message.getClass().toString(), actor.toString()); + LOG.debug("Sending remote message {} to {}", message.getClass().toString(), + actor.toString()); - Future future = - ask(actor, message, new Timeout(duration)); + Future future = ask(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" , e); } } @@ -192,15 +193,13 @@ public class ActorContext { * * @param actor the ActorSelection * @param message the message to send - * @param duration the maximum amount of time to send he message * @return a Future containing the eventual result */ - public Future executeRemoteOperationAsync(ActorSelection actor, Object message, - FiniteDuration duration) { + public Future executeRemoteOperationAsync(ActorSelection actor, Object message) { LOG.debug("Sending remote message {} to {}", message.getClass().toString(), actor.toString()); - return ask(actor, message, new Timeout(duration)); + return ask(actor, message, operationTimeout); } /** @@ -214,6 +213,13 @@ public class ActorContext { actor.tell(message, ActorRef.noSender()); } + public void sendShardOperationAsync(String shardName, Object message) { + ActorSelection primary = findPrimary(shardName); + + primary.tell(message, ActorRef.noSender()); + } + + /** * Execute an operation on the primary for a given shard *

@@ -223,16 +229,14 @@ public class ActorContext { * * @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) { + public Object executeShardOperation(String shardName, Object message) { ActorSelection primary = findPrimary(shardName); - return executeRemoteOperation(primary, message, duration); + return executeRemoteOperation(primary, message); } /** @@ -244,25 +248,47 @@ public class ActorContext { * * @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) { + public Object executeLocalShardOperation(String shardName, Object message) { ActorRef local = findLocalShard(shardName); if(local != null) { - return executeLocalOperation(local, message, duration); + return executeLocalOperation(local, message); } return null; } + /** + * 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); + } + + + public void shutdown() { shardManager.tell(PoisonPill.getInstance(), null); actorSystem.shutdown(); @@ -300,4 +326,22 @@ public class ActorContext { return clusterWrapper.getCurrentMemberName(); } + /** + * Send the message to each and every shard + * + * @param message + */ + 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 FiniteDuration getOperationDuration() { + return operationDuration; + } }