SPY-161: Cascade op cancellations.

Motivation
----------
When an operation is redistributed (be it because of reconnects
and/or "not my vbucket" responses, it gets cloned. The problem
is that the initial callback always points to the original
operation, not the cloned one. While the clone gets the callback,
the other way round (like cancelling from user-code) will never
trickle through to the cloned ops.

Modifications
-------------
In order to fix this issue, code has been added which connects
the original op with the cloned ones, by keeping a list of
cloned operations in the original op. To prevent overly long
stack in malicious scenarios, an overall clone-cap of 100
clones is introduced. If this cap is reached, the operation
will not be cloned further but cancelled.

Result
------
With this modifications, timeouts and cancellations from user code
will trickle through to the clone operations, prevent issues when
cloned operations are around "forever".

Change-Id: I304534a62202b705d78ca5f775f47afcd6be238a
Reviewed-on: http://review.couchbase.org/34883
Reviewed-by: Michael Nitschinger <michael.nitschinger@couchbase.com>
Tested-by: Michael Nitschinger <michael.nitschinger@couchbase.com>

Author: daschl

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

@@ -93,6 +93,12 @@ public class MemcachedConnection extends SpyThread {
    */
   private static final int EXCESSIVE_EMPTY = 0x1000000;
 
+  /**
+   * If an operation gets cloned more than this ceiling, cancel it for
+   * safety reasons.
+   */
+  private static final int MAX_CLONE_COUNT = 100;
+
   private static final String RECON_QUEUE_METRIC =
     "[MEM] Reconnecting Nodes (ReconnectQueue)";
   private static final String SHUTD_QUEUE_METRIC =
@@ -992,6 +998,13 @@ public void redistributeOperation(Operation op) {
       return;
     }
 
+    if (op.getCloneCount() >= MAX_CLONE_COUNT) {
+      getLogger().warn("Cancelling operation " + op + "because it has been "
+        + "retried (cloned) more than " + MAX_CLONE_COUNT + "times.");
+      op.cancel();
+      return;
+    }
+
     // The operation gets redistributed but has never been actually written,
     // it we just straight re-add it without cloning.
     if (op.getState() == OperationState.WRITE_QUEUED) {
@@ -1011,6 +1024,8 @@ public void redistributeOperation(Operation op) {
           KeyedOperation newKeyedOp = (KeyedOperation) newop;
           for (String k : newKeyedOp.getKeys()) {
             addOperation(k, newop);
+            op.addClone(newop);
+            newop.setCloneCount(op.getCloneCount()+1);
           }
         } else {
           newop.cancel();

src/main/java/net/spy/memcached/ops/Operation.java

@@ -165,4 +165,23 @@ public interface Operation {
    * @return the raw error message content.
    */
   byte[] getErrorMsg();
+
+  /**
+   * Add the clone from this operation.
+   *
+   * @param op the cloned operation.
+   */
+  void addClone(Operation op);
+
+  /**
+   * Returns the number of times this operation has been cloned.
+   *
+   * @return the number of clones.
+   */
+  int getCloneCount();
+
+  /**
+   * Sets the clone count for this operation.
+   */
+  void setCloneCount(int count);
 }

src/main/java/net/spy/memcached/protocol/BaseOperationImpl.java

@@ -25,8 +25,13 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
 
 import net.spy.memcached.MemcachedNode;
 import net.spy.memcached.compat.SpyObject;
@@ -65,6 +70,18 @@ public abstract class BaseOperationImpl extends SpyObject implements Operation {
       new HashSet<MemcachedNode>();
   private long writeCompleteTimestamp;
 
+  /**
+   * If the operation gets cloned, the reference is used to cascade cancellations
+   * and timeouts.
+   */
+  private List<Operation> clones =
+    Collections.synchronizedList(new ArrayList<Operation>());
+
+  /**
+   * Number of clones for this operation.
+   */
+  private volatile int cloneCount;
+
   public BaseOperationImpl() {
     super();
     creationTime = System.nanoTime();
@@ -98,6 +115,14 @@ public final OperationException getException() {
 
   public final synchronized void cancel() {
     cancelled = true;
+
+    synchronized (clones) {
+      Iterator<Operation> i = clones.iterator();
+      while(i.hasNext()) {
+        i.next().cancel();
+      }
+    }
+
     wasCancelled();
     callback.receivedStatus(CANCELLED);
     callback.complete();
@@ -193,6 +218,14 @@ public void setHandlingNode(MemcachedNode to) {
   @Override
   public synchronized void timeOut() {
     timedout = true;
+
+    synchronized (clones) {
+      Iterator<Operation> i = clones.iterator();
+      while(i.hasNext()) {
+        i.next().timeOut();
+      }
+    }
+
     callback.receivedStatus(TIMED_OUT);
     callback.complete();
   }
@@ -228,4 +261,18 @@ public long getWriteCompleteTimestamp() {
     return writeCompleteTimestamp;
   }
 
+  @Override
+  public void addClone(Operation op) {
+    clones.add(op);
+  }
+
+  @Override
+  public int getCloneCount() {
+    return cloneCount;
+  }
+
+  @Override
+  public void setCloneCount(int count) {
+    cloneCount = count;
+  }
 }