Ensure nodesMap updates are safe when topology changes.

This change much more closely (perhaps too defensively)
couples the Config and the list of nodes.  It also ensures
that the Map<String, MemcachedNode> is less prone to
concurrency problems and fixes a concurrency problem which
allowed nodes to be removed when receiving a not-my-vbucket
response.

This indicates we need to make some changes to how the entire
object model and config list are handled, but just to quickly
fix a race found, tie together the config and node list when
locating the right node.

Testing for this is rather manual at the moment.

Change-Id: Ida002fd1d510d20c432e77e01eefbc530d3b34b1
Reviewed-on: http://review.couchbase.org/8195
Tested-by: Matt Ingenthron <[email protected]>
Reviewed-by: Michael Wiederhold <[email protected]>
Reviewed-by: Matt Ingenthron <[email protected]>

Author: ingenthr

src/main/java/net/spy/memcached/vbucket/VBucketNodeLocator.java

@@ -2,24 +2,27 @@
 
 import java.net.InetSocketAddress;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
 
 import net.spy.memcached.MemcachedNode;
 import net.spy.memcached.NodeLocator;
 import net.spy.memcached.compat.SpyObject;
 import net.spy.memcached.vbucket.config.Config;
+import net.spy.memcached.vbucket.config.ConfigDifference;
 
 /**
  * Implementation of the {@link NodeLocator} interface that contains vbucket hashing methods
  */
 public class VBucketNodeLocator extends SpyObject implements NodeLocator {
 
-    private Map<String, MemcachedNode> nodesMap;
-
-    private Config config;
+    private final AtomicReference<TotalConfig> fullConfig;
 
     /**
      * Construct a VBucketNodeLocator over the given JSON configuration string.
@@ -29,23 +32,32 @@ public class VBucketNodeLocator extends SpyObject implements NodeLocator {
      */
     public VBucketNodeLocator(List<MemcachedNode> nodes, Config jsonConfig) {
         super();
-        setNodes(nodes);
-        setConfig(jsonConfig);
+        fullConfig = new AtomicReference<TotalConfig>();
+        fullConfig.set(new TotalConfig(jsonConfig, fillNodesEntries(nodes)));
     }
 
     /**
      * {@inheritDoc}
      */
     public MemcachedNode getPrimary(String k) {
+        TotalConfig totConfig = fullConfig.get();
+        Config config = totConfig.getConfig();
+        Map<String, MemcachedNode> nodesMap = totConfig.getNodesMap();
         int vbucket = config.getVbucketByKey(k);
         int serverNumber = config.getMaster(vbucket);
         String server = config.getServer(serverNumber);
-        // choose appropriate MemecachedNode according to config data
+        // choose appropriate MemcachedNode according to config data
         MemcachedNode pNode = nodesMap.get(server);
         if (pNode == null) {
-            getLogger().error("The node locator does not have a primary for key %s.", k);
+            getLogger().error("The node locator does not have a primary for key %s.  Wanted vbucket %s which should be on server %s.", k, vbucket, server);
+            getLogger().error("List of nodes has %s entries:", nodesMap.size());
+            Set<String> keySet = nodesMap.keySet();
+            Iterator<String> iterator = keySet.iterator();
+            while (iterator.hasNext()) {
+                String anode = iterator.next();
+                getLogger().error("MemcachedNode for %s is %s", anode, nodesMap.get(anode));
+            }
             Collection<MemcachedNode> nodes = nodesMap.values();
-            getLogger().error("MemcachedNode has %s entries:", nodesMap.size());
             for (MemcachedNode node : nodes) {
                 getLogger().error(node);
             }
@@ -58,14 +70,16 @@ public MemcachedNode getPrimary(String k) {
      * {@inheritDoc}
      */
     public Iterator<MemcachedNode> getSequence(String k) {
+        Map<String, MemcachedNode> nodesMap = fullConfig.get().getNodesMap();
         return nodesMap.values().iterator();
     }
 
     /**
      * {@inheritDoc}
      */
     public Collection<MemcachedNode> getAll() {
-        return this.nodesMap.values();
+	Map<String, MemcachedNode> nodesMap = fullConfig.get().getNodesMap();
+	return nodesMap.values();
     }
 
     /**
@@ -74,9 +88,17 @@ public Collection<MemcachedNode> getAll() {
     public NodeLocator getReadonlyCopy() {
         return this;
     }
-    public void updateLocator(final List<MemcachedNode> nodes, final Config conf) {
-        setNodes(nodes);
-        setConfig(conf);
+
+    public void updateLocator(final List<MemcachedNode> nodes, final Config newconf) {
+	// we'll get a new config for various reasons we don't care about, so check if we do care
+	Config current = fullConfig.get().getConfig();
+	ConfigDifference compareTo = current.compareTo(newconf);
+	if(compareTo.isSequenceChanged() || compareTo.getVbucketsChanges() > 0) {
+	    getLogger().debug("Updating configuration, received updated configuration with significant changes.");
+	    fullConfig.set(new TotalConfig(newconf, fillNodesEntries(nodes)));
+	} else {
+	    getLogger().debug("Received updated configuration with insignificant changes.");
+	}
     }
 
     /**
@@ -85,10 +107,11 @@ public void updateLocator(final List<MemcachedNode> nodes, final Config conf) {
      * @return vbucket index
      */
     public int getVBucketIndex(String key) {
-        return config.getVbucketByKey(key);
+        Config config = fullConfig.get().getConfig();
+	return config.getVbucketByKey(key);
     }
 
-    private void setNodes(Collection<MemcachedNode> nodes) {
+    private Map<String, MemcachedNode> fillNodesEntries(Collection<MemcachedNode> nodes) {
         HashMap<String, MemcachedNode> vbnodesMap = new HashMap<String, MemcachedNode>();
         getLogger().debug("Updating nodesMap in VBucketNodeLocator.");
         for (MemcachedNode node : nodes) {
@@ -101,12 +124,9 @@ private void setNodes(Collection<MemcachedNode> nodes) {
 	    vbnodesMap.put(hostname, node);
         }
 
-        this.nodesMap = vbnodesMap;
+        return Collections.unmodifiableMap(vbnodesMap);
     }
 
-    private void setConfig(final Config config) {
-        this.config = config;
-    }
 
     /**
      * Method returns the node that is not contained in the specified collection of the failed nodes
@@ -115,6 +135,8 @@ private void setConfig(final Config config) {
      * @return The first MemcachedNode which meets requirements
      */
     public MemcachedNode getAlternative(String k, Collection<MemcachedNode> notMyVbucketNodes) {
+        // it's safe to only copy the map here, only removing references found to be incorrect, and trying remaining
+        Map<String, MemcachedNode> nodesMap = new HashMap<String,MemcachedNode>(fullConfig.get().getNodesMap());
         Collection<MemcachedNode> nodes = nodesMap.values();
         nodes.removeAll(notMyVbucketNodes);
         if (nodes.isEmpty()) {
@@ -123,4 +145,23 @@ public MemcachedNode getAlternative(String k, Collection<MemcachedNode> notMyVbu
             return nodes.iterator().next();
         }
     }
+
+    private class TotalConfig {
+        private Config config;
+        private Map<String, MemcachedNode> nodesMap;
+
+        public TotalConfig(Config newConfig, Map<String, MemcachedNode> newMap) {
+	    config = newConfig;
+	    nodesMap = Collections.unmodifiableMap(newMap);
+        }
+
+        protected Config getConfig() {
+	    return config;
+        }
+
+        protected Map<String, MemcachedNode> getNodesMap() {
+	    return nodesMap;
+        }
+    }
+
 }