SPY-132: Adding Callbacks to Futures.

This changeset adds a onComplete callback to futures and executes
the callback in a configurable ExecutorService. By default, this
is a fixed size thread pool (threads = num of processors), but
configurable through the Factory.

Change-Id: I516af74918e57521542a0bad2ff7142b75ab7b13
Reviewed-on: http://review.couchbase.org/28025
Tested-by: Michael Nitschinger <[email protected]>
Reviewed-by: Matt Ingenthron <[email protected]>

Author: daschl

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

@@ -30,6 +30,7 @@
 import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
 
 import net.spy.memcached.auth.AuthDescriptor;
 import net.spy.memcached.metrics.MetricCollector;
@@ -82,6 +83,12 @@ MemcachedNode createMemcachedNode(SocketAddress sa, SocketChannel c,
    */
   long getOpQueueMaxBlockTime();
 
+  /**
+   * Get the ExecutorService which is used to asynchronously execute listeners
+   * on futures.
+   */
+  ExecutorService getListenerExecutorService();
+
   /**
    * Create a NodeLocator instance for the given list of nodes.
    */

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

@@ -32,6 +32,8 @@
 import java.util.List;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 
@@ -114,6 +116,12 @@ public class DefaultConnectionFactory extends SpyObject implements
    */
   public static final MetricType DEFAULT_METRIC_TYPE = MetricType.OFF;
 
+  /**
+   * The ExecutorService in which the listener callbacks will be executed.
+   */
+  public static final ExecutorService DEFAULT_LISTENER_EXECUTOR_SERVICE =
+    Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
+
   protected final int opQueueLen;
   private final int readBufSize;
   private final HashAlgorithm hashAlg;
@@ -249,6 +257,10 @@ public long getOpQueueMaxBlockTime() {
     return DEFAULT_OP_QUEUE_MAX_BLOCK_TIME;
   }
 
+  public ExecutorService getListenerExecutorService() {
+    return DEFAULT_LISTENER_EXECUTOR_SERVICE;
+  }
+
   /*
    * (non-Javadoc)
    *

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

@@ -1,6 +1,6 @@
 /**
  * Copyright (C) 2006-2009 Dustin Sallings
- * Copyright (C) 2009-2011 Couchbase, Inc.
+ * Copyright (C) 2009-2013 Couchbase, Inc.
  *
  * Permission is hereby granted, free of charge, to any person obtaining a copy
  * of this software and associated documentation files (the "Software"), to deal
@@ -41,6 +41,7 @@
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -154,6 +155,8 @@ public class MemcachedClient extends SpyObject implements MemcachedClientIF,
 
   protected final AuthThreadMonitor authMonitor = new AuthThreadMonitor();
 
+  protected final ExecutorService executorService;
+
   /**
    * Get a memcache client operating on the specified memcached locations.
    *
@@ -205,6 +208,7 @@ public MemcachedClient(ConnectionFactory cf, List<InetSocketAddress> addrs)
     assert mconn != null : "Connection factory failed to make a connection";
     operationTimeout = cf.getOperationTimeout();
     authDescriptor = cf.getAuthDescriptor();
+    executorService = cf.getListenerExecutorService();
     if (authDescriptor != null) {
       addObserver(this);
     }
@@ -292,7 +296,8 @@ private <T> OperationFuture<Boolean> asyncStore(StoreType storeType,
     CachedData co = tc.encode(value);
     final CountDownLatch latch = new CountDownLatch(1);
     final OperationFuture<Boolean> rv =
-        new OperationFuture<Boolean>(key, latch, operationTimeout);
+      new OperationFuture<Boolean>(key, latch, operationTimeout,
+      executorService);
     Operation op = opFact.store(storeType, key, co.getFlags(), exp,
         co.getData(), new StoreOperation.Callback() {
             public void receivedStatus(OperationStatus val) {
@@ -321,7 +326,7 @@ private <T> OperationFuture<Boolean> asyncCat(ConcatenationType catType,
     CachedData co = tc.encode(value);
     final CountDownLatch latch = new CountDownLatch(1);
     final OperationFuture<Boolean> rv = new OperationFuture<Boolean>(key,
-        latch, operationTimeout);
+        latch, operationTimeout, executorService);
     Operation op = opFact.cat(catType, cas, key, co.getData(),
         new OperationCallback() {
           public void receivedStatus(OperationStatus val) {
@@ -367,7 +372,8 @@ public <T> OperationFuture<Boolean> touch(final String key, final int exp,
       final Transcoder<T> tc) {
     final CountDownLatch latch = new CountDownLatch(1);
     final OperationFuture<Boolean> rv =
-        new OperationFuture<Boolean>(key, latch, operationTimeout);
+      new OperationFuture<Boolean>(key, latch, operationTimeout,
+      executorService);
 
     Operation op = opFact.touch(key, exp, new OperationCallback() {
       public void receivedStatus(OperationStatus status) {
@@ -587,7 +593,8 @@ public <T> OperationFuture<Boolean> prepend(String key, T val,
     CachedData co = tc.encode(value);
     final CountDownLatch latch = new CountDownLatch(1);
     final OperationFuture<CASResponse> rv =
-      new OperationFuture<CASResponse>(key, latch, operationTimeout);
+      new OperationFuture<CASResponse>(key, latch, operationTimeout,
+      executorService);
     Operation op = opFact.cas(StoreType.set, key, casId, co.getFlags(), exp,
         co.getData(), new StoreOperation.Callback() {
             public void receivedStatus(OperationStatus val) {
@@ -947,7 +954,8 @@ public OperationFuture<Boolean> replace(String key, int exp, Object o) {
   public <T> GetFuture<T> asyncGet(final String key, final Transcoder<T> tc) {
 
     final CountDownLatch latch = new CountDownLatch(1);
-    final GetFuture<T> rv = new GetFuture<T>(latch, operationTimeout, key);
+    final GetFuture<T> rv = new GetFuture<T>(latch, operationTimeout, key,
+      executorService);
     Operation op = opFact.get(key, new GetOperation.Callback() {
       private Future<T> val = null;
 
@@ -997,7 +1005,8 @@ public <T> OperationFuture<CASValue<T>> asyncGets(final String key,
 
     final CountDownLatch latch = new CountDownLatch(1);
     final OperationFuture<CASValue<T>> rv =
-        new OperationFuture<CASValue<T>>(key, latch, operationTimeout);
+      new OperationFuture<CASValue<T>>(key, latch, operationTimeout,
+      executorService);
 
     Operation op = opFact.gets(key, new GetsOperation.Callback() {
       private CASValue<T> val = null;
@@ -1228,7 +1237,7 @@ public <T> BulkFuture<Map<String, T>> asyncGetBulk(Iterator<String> keyIter,
 
     final CountDownLatch latch = new CountDownLatch(chunks.size());
     final Collection<Operation> ops = new ArrayList<Operation>(chunks.size());
-    final BulkGetFuture<T> rv = new BulkGetFuture<T>(m, ops, latch);
+    final BulkGetFuture<T> rv = new BulkGetFuture<T>(m, ops, latch, executorService);
 
     GetOperation.Callback cb = new GetOperation.Callback() {
       @SuppressWarnings("synthetic-access")
@@ -1396,7 +1405,7 @@ public <T> OperationFuture<CASValue<T>> asyncGetAndTouch(final String key,
       final int exp, final Transcoder<T> tc) {
     final CountDownLatch latch = new CountDownLatch(1);
     final OperationFuture<CASValue<T>> rv = new OperationFuture<CASValue<T>>(
-        key, latch, operationTimeout);
+        key, latch, operationTimeout, executorService);
 
     Operation op = opFact.getAndTouch(key, exp,
         new GetAndTouchOperation.Callback() {
@@ -1843,7 +1852,7 @@ private OperationFuture<Long> asyncMutate(Mutator m, String key, long by,
       long def, int exp) {
     final CountDownLatch latch = new CountDownLatch(1);
     final OperationFuture<Long> rv =
-        new OperationFuture<Long>(key, latch, operationTimeout);
+        new OperationFuture<Long>(key, latch, operationTimeout, executorService);
     Operation op = opFact.mutate(m, key, by, def, exp,
         new OperationCallback() {
           public void receivedStatus(OperationStatus s) {
@@ -2023,7 +2032,7 @@ public OperationFuture<Boolean> delete(String key) {
   public OperationFuture<Boolean> delete(String key, long cas) {
     final CountDownLatch latch = new CountDownLatch(1);
     final OperationFuture<Boolean> rv = new OperationFuture<Boolean>(key,
-        latch, operationTimeout);
+        latch, operationTimeout, executorService);
 
     DeleteOperation.Callback callback = new DeleteOperation.Callback() {
       public void receivedStatus(OperationStatus s) {
@@ -2082,7 +2091,7 @@ public void complete() {
     });
 
     return new OperationFuture<Boolean>(null, blatch, flushResult,
-        operationTimeout) {
+        operationTimeout, executorService) {
       @Override
       public boolean cancel(boolean ign) {
         boolean rv = false;

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

@@ -49,6 +49,7 @@
 import java.util.TreeMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 
 import net.spy.memcached.compat.SpyThread;
@@ -131,6 +132,7 @@ public class MemcachedConnection extends SpyThread {
   private final Collection<Operation> retryOps;
   protected final ConcurrentLinkedQueue<MemcachedNode> nodesToShutdown;
   private final boolean verifyAliveOnConnect;
+  private final ExecutorService listenerExecutorService;
 
   protected final MetricCollector metrics;
   protected final MetricType metricType;
@@ -158,6 +160,7 @@ public MemcachedConnection(int bufSize, ConnectionFactory f,
     selector = Selector.open();
     retryOps = new ArrayList<Operation>();
     nodesToShutdown = new ConcurrentLinkedQueue<MemcachedNode>();
+    listenerExecutorService = f.getListenerExecutorService();
     this.bufSize = bufSize;
     this.connectionFactory = f;
 
@@ -485,7 +488,8 @@ private void handleIO(SelectionKey sk) {
             // Test to see if it's truly alive, could be a hung process, OS
             final CountDownLatch latch = new CountDownLatch(1);
             final OperationFuture<Boolean> rv =
-              new OperationFuture<Boolean>("noop", latch, 2500);
+              new OperationFuture<Boolean>("noop", latch, 2500,
+                listenerExecutorService);
             NoopOperation testOp = opFact.noop(new OperationCallback() {
               public void receivedStatus(OperationStatus status) {
                 rv.set(status.isSuccess(), status);