[2/2] incubator-geode git commit: async close pool now sets max threads and has an unlimited queue. Also a wait time can now be enabled.

[ds...@apache.org]

async close pool now sets max threads and has an unlimited queue.
Also a wait time can now be enabled.


Project: http://git-wip-us.apache.org/repos/asf/incubator-geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-geode/commit/12a36ca7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-geode/tree/12a36ca7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-geode/diff/12a36ca7

Branch: refs/heads/feature/GEODE-332
Commit: 12a36ca70a10694d717f7d9b82f61009c08cd2b3
Parents: f64c166
Author: Darrel Schneider <ds...@pivotal.io>
Authored: Wed Sep 16 11:13:48 2015 -0700
Committer: Darrel Schneider <ds...@pivotal.io>
Committed: Wed Sep 16 11:13:48 2015 -0700

----------------------------------------------------------------------
 .../gemfire/internal/SocketCreator.java         | 33 ++++++++++++++++----
 1 file changed, 27 insertions(+), 6 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/12a36ca7/gemfire-core/src/main/java/com/gemstone/gemfire/internal/SocketCreator.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/SocketCreator.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/SocketCreator.java
index 0688c3d..dad5788 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/SocketCreator.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/SocketCreator.java
@@ -82,10 +82,13 @@ import com.gemstone.org.jgroups.util.ConnectionWatcher;
 
 import java.util.*;
 import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 import javax.net.ssl.*;
 
@@ -1206,6 +1209,10 @@ public class SocketCreator  implements com.gemstone.org.jgroups.util.SockCreator
   private static ThreadPoolExecutor asyncCloseExecutor;
   /** Number of seconds to wait before timing out an unused async close thread. Default is 120 (2 minutes). */
   private final static long ASYNC_CLOSE_POOL_KEEP_ALIVE_TIME = Long.getLong("p2p.ASYNC_CLOSE_POOL_KEEP_ALIVE_TIME", 120).longValue();
+  /** Maximum number of threads that can be doing a socket close. Any close requests over this max will queue up waiting for a thread. */
+  private final static int ASYNC_CLOSE_POOL_MAX_THREADS = Integer.getInteger("p2p.ASYNC_CLOSE_POOL_MAX_THREADS", 32).intValue();
+  /** How many milliseconds the synchronous requester waits for the async close to happen. Default is 0. Prior releases waited 50ms. */ 
+  private final static long ASYNC_CLOSE_WAIT_MILLISECONDS = Long.getLong("p2p.ASYNC_CLOSE_WAIT_MILLISECONDS", 0).longValue();
   
   private static synchronized ThreadPoolExecutor getAsyncThreadExecutor() {
     ThreadPoolExecutor pool = asyncCloseExecutor;
@@ -1218,8 +1225,8 @@ public class SocketCreator  implements com.gemstone.org.jgroups.util.SockCreator
           return thread;
         } 
       }; 
-      BlockingQueue synchronousQueue = new SynchronousQueue(); 
-      pool = new ThreadPoolExecutor(1, Integer.MAX_VALUE, ASYNC_CLOSE_POOL_KEEP_ALIVE_TIME, TimeUnit.SECONDS, synchronousQueue, tf);
+      BlockingQueue workQueue = new LinkedBlockingQueue(); 
+      pool = new ThreadPoolExecutor(1, ASYNC_CLOSE_POOL_MAX_THREADS, ASYNC_CLOSE_POOL_KEEP_ALIVE_TIME, TimeUnit.SECONDS, workQueue, tf);
       asyncCloseExecutor = pool;
     }
     return pool;
@@ -1232,9 +1239,23 @@ public class SocketCreator  implements com.gemstone.org.jgroups.util.SockCreator
     }
   }
   private static synchronized void asyncExecute(Runnable r) {
-    // The old code waited 50ms for the async task to complete.
-    // Should this code use submit on the executor and also wait 50ms?
-    getAsyncThreadExecutor().execute(r);
+    // Waiting 50ms for the async close request to complete is what the old (close per thread)
+    // code did. But now that we will not create a thread for every close request
+    // it seems better to let the thread that requested the close to move on quickly.
+    // So the default has changed to not wait. The system property p2p.ASYNC_CLOSE_WAIT_MILLISECONDS
+    // can be set to how many milliseconds to wait.
+    if (ASYNC_CLOSE_WAIT_MILLISECONDS == 0) {
+      getAsyncThreadExecutor().execute(r);
+    } else {
+      Future future = getAsyncThreadExecutor().submit(r);
+      try {
+        future.get(ASYNC_CLOSE_WAIT_MILLISECONDS, TimeUnit.MILLISECONDS);
+      } catch (InterruptedException | ExecutionException | TimeoutException e) {
+        // We want this code to wait at most 50ms for the close to happen.
+        // It is ok to ignore these exception and let the close continue
+        // in the background.
+      }
+    }
   }
   /**
    * Closes the specified socket in a background thread and waits a limited