INTERNAL: Using addOperations for atomicity in Broadcast Op.

Author: brido4125

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

@@ -1901,6 +1901,7 @@ public OperationFuture<Boolean> flush(final String prefix, final int delay) {
     Collection<MemcachedNode> nodes = getAllNodes();
     final BroadcastFuture<Boolean> rv
             = new BroadcastFuture<Boolean>(operationTimeout, Boolean.TRUE, nodes.size());
+    final Map<MemcachedNode, Operation> opsMap = new HashMap<MemcachedNode, Operation>();
 
     checkState();
     for (MemcachedNode node : nodes) {
@@ -1917,9 +1918,10 @@ public void complete() {
           rv.complete();
         }
       });
-      rv.addOp(op);
-      getMemcachedConnection().addOperation(node, op);
+      opsMap.put(node, op);
     }
+    rv.addOperations(opsMap.values());
+    getMemcachedConnection().addOperations(opsMap);
     return rv;
   }
 

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

@@ -1549,6 +1549,7 @@ public Map<SocketAddress, String> getVersions() {
     final BroadcastFuture<Map<SocketAddress, String>> future
             = new BroadcastFuture<Map<SocketAddress, String>>(
                     operationTimeout, result, nodes.size());
+    final Map<MemcachedNode, Operation> opsMap = new HashMap<MemcachedNode, Operation>();
 
     checkState();
     for (MemcachedNode node : nodes) {
@@ -1564,9 +1565,10 @@ public void complete() {
           future.complete();
         }
       });
-      future.addOp(op);
-      conn.addOperation(node, op);
+      opsMap.put(node, op);
     }
+    future.addOperations(opsMap.values());
+    conn.addOperations(opsMap);
 
     Map<SocketAddress, String> rv = null;
     try {
@@ -1612,6 +1614,7 @@ public Map<SocketAddress, Map<String, String>> getStats(final String arg) {
     final BroadcastFuture<Map<SocketAddress, Map<String, String>>> future
             = new BroadcastFuture<Map<SocketAddress, Map<String, String>>>(
                     operationTimeout, resultMap, nodes.size());
+    final Map<MemcachedNode, Operation> opsMap = new HashMap<MemcachedNode, Operation>();
 
     checkState();
     for (MemcachedNode node : nodes) {
@@ -1637,9 +1640,10 @@ public void complete() {
           future.complete();
         }
       });
-      future.addOp(op);
-      conn.addOperation(node, op);
+      opsMap.put(node, op);
     }
+    future.addOperations(opsMap.values());
+    conn.addOperations(opsMap);
 
     Map<SocketAddress, Map<String, String>> rv = null;
     try {
@@ -1965,6 +1969,7 @@ public Future<Boolean> flush(final int delay) {
     Collection<MemcachedNode> nodes = getAllNodes();
     final BroadcastFuture<Boolean> rv
             = new BroadcastFuture<Boolean>(operationTimeout, Boolean.TRUE, nodes.size());
+    final Map<MemcachedNode, Operation> opsMap = new HashMap<MemcachedNode, Operation>();
 
     checkState();
     for (MemcachedNode node : nodes) {
@@ -1981,9 +1986,10 @@ public void complete() {
           rv.complete();
         }
       });
-      rv.addOp(op);
-      conn.addOperation(node, op);
+      opsMap.put(node, op);
     }
+    rv.addOperations(opsMap.values());
+    conn.addOperations(opsMap);
     return rv;
   }
 
@@ -2005,6 +2011,7 @@ public Set<String> listSaslMechanisms() {
     final BroadcastFuture<ConcurrentMap<String, String>> future
             = new BroadcastFuture<ConcurrentMap<String, String>>(
                     operationTimeout, resultMap, nodes.size());
+    final Map<MemcachedNode, Operation> opsMap = new HashMap<MemcachedNode, Operation>();
 
     checkState();
     for (MemcachedNode node : nodes) {
@@ -2021,9 +2028,10 @@ public void complete() {
           future.complete();
         }
       });
-      future.addOp(op);
-      conn.addOperation(node, op);
+      opsMap.put(node, op);
     }
+    future.addOperations(opsMap.values());
+    conn.addOperations(opsMap);
 
     Set<String> rv = null;
     try {

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

@@ -1482,13 +1482,8 @@ public void addOperation(final MemcachedNode node, final Operation o) {
 
   public void addOperations(final Map<MemcachedNode, Operation> ops) {
     for (Map.Entry<MemcachedNode, Operation> me : ops.entrySet()) {
-      final MemcachedNode node = me.getKey();
-      Operation o = me.getValue();
-      node.addOpToInputQ(o);
-      addedQueue.offer(node);
+      addOperation(me.getKey(), me.getValue());
     }
-    Selector s = selector.wakeup();
-    assert s == selector : "Wakeup returned the wrong selector.";
   }
 
   public void wakeUpSelector() {

src/main/java/net/spy/memcached/internal/BroadcastFuture.java

@@ -84,8 +84,8 @@ public T get(long duration, TimeUnit units)
     return objRef.get();
   }
 
-  public void addOp(Operation op) {
-    ops.add(op);
+  public void addOperations(Collection<Operation> ops) {
+    this.ops.addAll(ops);
   }
 
   public void complete() {