SPY-194: Allow to optionally bound retry queue.

Motivation
----------
If the retry queue might grow very large this can lead to
uncontrollable side effects (like heap growth), since this
queue is not accessible from the outside of spymemcached.

Modifications
-------------
This change adds the system property "net.spy.retryQueueSize" which
allows to optionally bound the retry queue to the given size.

Note that to maximize backwards compatibility and to not introduce
any regressions the actual queue has not been changed but rather
the code which inserts into this queue is now checking for the
limit.

If the queue is full the operation is cancelled instead of sending
it into the retry queue.

Result
------
It is now possible to bound the retry queue.

Change-Id: I86452ef8dcddcf2c4acfc065fd9db94510665ac3
Reviewed-on: http://review.couchbase.org/69938
Tested-by: Michael Nitschinger <michael@nitschinger.at>
Reviewed-by: Sergey Avseyev <sergey.avseyev@gmail.com>

Author: daschl

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

@@ -97,6 +97,11 @@ public class MemcachedConnection extends SpyThread {
    */
   private static final int DEFAULT_WAKEUP_DELAY = 1000;
 
+  /**
+   * By default, do not bound the retry queue.
+   */
+  private static final int DEFAULT_RETRY_QUEUE_SIZE = -1;
+
   /**
    * If an operation gets cloned more than this ceiling, cancel it for
    * safety reasons.
@@ -240,6 +245,11 @@ public class MemcachedConnection extends SpyThread {
    */
   private final int wakeupDelay;
 
+  /**
+   * Optionally bound the retry queue if set via system property.
+   */
+  private final int retryQueueSize;
+
   /**
    * Construct a {@link MemcachedConnection}.
    *
@@ -279,6 +289,10 @@ public MemcachedConnection(final int bufSize, final ConnectionFactory f,
     wakeupDelay = Integer.parseInt( System.getProperty("net.spy.wakeupDelay",
       Integer.toString(DEFAULT_WAKEUP_DELAY)));
 
+    retryQueueSize = Integer.parseInt(System.getProperty("net.spy.retryQueueSize",
+        Integer.toString(DEFAULT_RETRY_QUEUE_SIZE)));
+    getLogger().info("Setting retryQueueSize to " + retryQueueSize);
+
     List<MemcachedNode> connections = createConnections(a);
     locator = f.createLocator(connections);
 
@@ -878,7 +892,7 @@ private void readBufferAndLogMetrics(final Operation currentOp,
       assert op == currentOp : "Expected to pop " + currentOp + " got "
         + op;
 
-      retryOps.add(currentOp);
+      retryOperation(currentOp);
       metrics.markMeter(OVERALL_RESPONSE_RETRY_METRIC);
     }
   }
@@ -1484,9 +1498,18 @@ public boolean isShutDown() {
   /**
    * Add a operation to the retry queue.
    *
+   * If the retry queue size is bounded and the size of the queue is reaching
+   * that boundary, the operation is cancelled rather than added to the
+   * retry queue.
+   *
    * @param op the operation to retry.
    */
   public void retryOperation(Operation op) {
+    if (retryQueueSize >= 0 && retryOps.size() >= retryQueueSize) {
+      if (!op.isCancelled()) {
+        op.cancel();
+      }
+    }
     retryOps.add(op);
   }