Bug 2038: Ensure only one concurrent 3-phase commit in Shard

[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / ShardManager.java

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardManager.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardManager.java
index 5874c5296f0ebd8d1b5085abd3797f6c77907618..e68628dbf5c1fc992afb30ae986fff8ed8f6eef1 100644 (file)
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardManager.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardManager.java
@@ -24,11 +24,13 @@ import akka.persistence.RecoveryCompleted;
 import akka.persistence.RecoveryFailure;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
-import org.opendaylight.controller.cluster.common.actor.AbstractUntypedPersistentActor;
+import org.opendaylight.controller.cluster.common.actor.AbstractUntypedPersistentActorWithMetering;
 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
 import org.opendaylight.controller.cluster.datastore.identifiers.ShardManagerIdentifier;
 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shardmanager.ShardManagerInfo;
 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shardmanager.ShardManagerInfoMBean;
+import org.opendaylight.controller.cluster.datastore.messages.ActorInitialized;
+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;
@@ -37,7 +39,6 @@ import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolve
 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.controller.cluster.datastore.utils.ActorContext;
 import org.opendaylight.yangtools.yang.model.api.ModuleIdentifier;
 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
 import scala.concurrent.duration.Duration;
@@ -60,7 +61,7 @@ import java.util.Set;
  * <li> Monitor the cluster members and store their addresses
  * <ul>
  */
-public class ShardManager extends AbstractUntypedPersistentActor {
+public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
 
     protected final LoggingAdapter LOG =
         Logging.getLogger(getContext().system(), this);
@@ -128,6 +129,8 @@ public class ShardManager extends AbstractUntypedPersistentActor {
             findLocalShard((FindLocalShard) message);
         } else if (message instanceof UpdateSchemaContext) {
             updateSchemaContext(message);
+        } else if(message instanceof ActorInitialized) {
+            onActorInitialized(message);
         } else if (message instanceof ClusterEvent.MemberUp){
             memberUp((ClusterEvent.MemberUp) message);
         } else if(message instanceof ClusterEvent.MemberRemoved) {
@@ -140,6 +143,31 @@ public class ShardManager extends AbstractUntypedPersistentActor {
 
     }
 
+    private void onActorInitialized(Object message) {
+        final ActorRef sender = getSender();
+
+        if (sender == null) {
+            return; //why is a non-actor sending this message? Just ignore.
+        }
+
+        String actorName = sender.path().name();
+        //find shard name from actor name; actor name is stringified shardId
+        ShardIdentifier shardId = ShardIdentifier.builder().fromShardIdString(actorName).build();
+
+        if (shardId.getShardName() == null) {
+            return;
+        }
+        markShardAsInitialized(shardId.getShardName());
+    }
+
+    @VisibleForTesting protected void markShardAsInitialized(String shardName) {
+        LOG.debug("Initializing shard [{}]", shardName);
+        ShardInformation shardInformation = localShards.get(shardName);
+        if (shardInformation != null) {
+            shardInformation.setShardInitialized(true);
+        }
+    }
+
     @Override protected void handleRecover(Object message) throws Exception {
 
         if(message instanceof SchemaContextModules){
@@ -158,16 +186,23 @@ public class ShardManager extends AbstractUntypedPersistentActor {
     }
 
     private void findLocalShard(FindLocalShard message) {
-        ShardInformation shardInformation =
-            localShards.get(message.getShardName());
+        ShardInformation shardInformation = localShards.get(message.getShardName());
 
-        if(shardInformation != null){
-            getSender().tell(new LocalShardFound(shardInformation.getActor()), getSelf());
+        if(shardInformation == null){
+            getSender().tell(new LocalShardNotFound(message.getShardName()), getSelf());
             return;
         }
 
-        getSender().tell(new LocalShardNotFound(message.getShardName()),
-            getSelf());
+        sendResponse(shardInformation, new LocalShardFound(shardInformation.getActor()));
+    }
+
+    private void sendResponse(ShardInformation shardInformation, Object message) {
+        if (!shardInformation.isShardInitialized()) {
+            getSender().tell(new ActorNotInitialized(), getSelf());
+            return;
+        }
+
+        getSender().tell(message, getSelf());
     }
 
     private void memberRemoved(ClusterEvent.MemberRemoved message) {
@@ -177,7 +212,7 @@ public class ShardManager extends AbstractUntypedPersistentActor {
     private void memberUp(ClusterEvent.MemberUp message) {
         String memberName = message.member().roles().head();
 
-        memberNameToAddress.put(memberName , message.member().address());
+        memberNameToAddress.put(memberName, message.member().address());
 
         for(ShardInformation info : localShards.values()){
             String shardName = info.getShardName();
@@ -230,28 +265,27 @@ public class ShardManager extends AbstractUntypedPersistentActor {
     }
 
     private void findPrimary(FindPrimary message) {
+        final ActorRef sender = getSender();
         String shardName = message.getShardName();
 
         // First see if the there is a local replica for the shard
         ShardInformation info = localShards.get(shardName);
-        if(info != null) {
+        if (info != null) {
             ActorPath shardPath = info.getActorPath();
-            if (shardPath != null) {
-                getSender()
-                    .tell(
-                        new PrimaryFound(shardPath.toString()).toSerializable(),
-                        getSelf());
-                return;
-            }
+            sendResponse(info, new PrimaryFound(shardPath.toString()).toSerializable());
+            return;
         }
 
-        List<String> members =
-            configuration.getMembersFromShardName(shardName);
+        List<String> members = configuration.getMembersFromShardName(shardName);
 
         if(cluster.getCurrentMemberName() != null) {
             members.remove(cluster.getCurrentMemberName());
         }
 
+        /**
+         * FIXME: Instead of sending remote shard actor path back to sender,
+         * forward FindPrimary message to remote shard manager
+         */
         // There is no way for us to figure out the primary (for now) so assume
         // that one of the remote nodes is a primary
         for(String memberName : members) {
@@ -307,8 +341,8 @@ public class ShardManager extends AbstractUntypedPersistentActor {
             ShardIdentifier shardId = getShardIdentifier(memberName, shardName);
             Map<ShardIdentifier, String> peerAddresses = getPeerAddresses(shardName);
             ActorRef actor = getContext()
-                .actorOf(Shard.props(shardId, peerAddresses, datastoreContext, schemaContext).
-                    withMailbox(ActorContext.MAILBOX), shardId.toString());
+                .actorOf(Shard.props(shardId, peerAddresses, datastoreContext, schemaContext),
+                    shardId.toString());
             localShardActorNames.add(shardId.toString());
             localShards.put(shardName, new ShardInformation(shardName, actor, peerAddresses));
         }
@@ -377,6 +411,7 @@ public class ShardManager extends AbstractUntypedPersistentActor {
         private final ActorRef actor;
         private final ActorPath actorPath;
         private final Map<ShardIdentifier, String> peerAddresses;
+        private boolean shardInitialized = false; //flag that determines if the actor is ready for business
 
         private ShardInformation(String shardName, ActorRef actor,
             Map<ShardIdentifier, String> peerAddresses) {
@@ -414,6 +449,14 @@ public class ShardManager extends AbstractUntypedPersistentActor {
 
             }
         }
+
+        public boolean isShardInitialized() {
+            return shardInitialized;
+        }
+
+        public void setShardInitialized(boolean shardInitialized) {
+            this.shardInitialized = shardInitialized;
+        }
     }
 
     private static class ShardManagerCreator implements Creator<ShardManager> {