ENHANCE: Moved cache list change logic from IO thread to CacheManger thread.

Author: brido4125

src/main/java/net/spy/memcached/CacheManager.java

@@ -398,6 +398,20 @@ public void run() {
             waitBeforeRetryMonitorCheck(1000L - elapsed);
           }
         }
+
+        // Only perform cache node list updates when connectivity with ZK is normal.
+        if (!isDead) {
+          // Handling cacheList changes to a MemcachedConnection resulting
+          // from the processing of a CacheMonitor's Watch event.
+          for (ArcusClient ac : getAC()) {
+            try {
+              ac.getMemcachedConnection().handleCacheNodesChange();
+            } catch (IOException e) {
+              getLogger().error("Cache List update error in ArcusClient {}, exception = {}",
+                      ac.getName(), e.getCause());
+            }
+          }
+        }
       }
     }
     getLogger().info("Close cache manager.");

src/main/java/net/spy/memcached/MemcachedConnection.java

@@ -44,6 +44,7 @@
 import java.util.SortedMap;
 import java.util.TreeMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -142,7 +143,7 @@ public MemcachedConnection(String name, ConnectionFactory f,
     timeoutRatioThreshold = f.getTimeoutRatioThreshold();
     timeoutDurationThreshold = f.getTimeoutDurationThreshold();
     selector = Selector.open();
-    List<MemcachedNode> connections = new ArrayList<MemcachedNode>(a.size());
+    List<MemcachedNode> connections = new CopyOnWriteArrayList<MemcachedNode>();
     for (SocketAddress sa : a) {
       connections.add(makeMemcachedNode(connName, sa));
     }
@@ -304,9 +305,6 @@ public void handleIO() throws IOException {
     }
     /* ENABLE_REPLICATION end */
 
-    // Deal with the memcached server group that's been added by CacheManager.
-    handleCacheNodesChange();
-
     if (!reconnectQueue.isEmpty()) {
       attemptReconnects();
     }
@@ -735,7 +733,7 @@ public void setCacheNodesChange(String addrs) {
       alterNodesChange.set(old);
     }
     /* ENABLE_MIGRATION end */
-    selector.wakeup();
+    wakeUpSelector();
   }
 
   /* ENABLE_MIGRATION if */