Github user kohlmu-pivotal commented on a diff in the pull request:
https://github.com/apache/geode/pull/702#discussion_r133255222
--- Diff:
geode-core/src/main/java/org/apache/geode/internal/net/SocketCloser.java ---
@@ -96,46 +99,55 @@ public int getMaxThreads() {
return this.asyncClosePoolMaxThreads;
}
- private ThreadPoolExecutor getAsyncThreadExecutor(String address) {
- synchronized (asyncCloseExecutors) {
- ThreadPoolExecutor pool = asyncCloseExecutors.get(address);
- if (pool == null) {
- final ThreadGroup tg =
LoggingThreadGroup.createThreadGroup("Socket asyncClose", logger);
- ThreadFactory tf = new ThreadFactory() {
- public Thread newThread(final Runnable command) {
- Thread thread = new Thread(tg, command);
- thread.setDaemon(true);
- return thread;
- }
- };
- BlockingQueue<Runnable> workQueue = new
LinkedBlockingQueue<Runnable>();
- pool = new ThreadPoolExecutor(this.asyncClosePoolMaxThreads,
this.asyncClosePoolMaxThreads,
- this.asyncClosePoolKeepAliveSeconds, TimeUnit.SECONDS,
workQueue, tf);
- pool.allowCoreThreadTimeOut(true);
- asyncCloseExecutors.put(address, pool);
+ private ExecutorService getAsyncThreadExecutor(String address) {
+ ExecutorService executorService = asyncCloseExecutors.get(address);
+ if (executorService == null) {
+ // To be used for pre-1.8 jdk releases.
+ // createThreadPool();
+
+ executorService =
Executors.newWorkStealingPool(asyncClosePoolMaxThreads);
+
+ ExecutorService previousThreadPoolExecutor =
+ asyncCloseExecutors.putIfAbsent(address, executorService);
+
+ if (previousThreadPoolExecutor != null) {
+ executorService.shutdownNow();
+ return previousThreadPoolExecutor;
}
- return pool;
}
+ return executorService;
+ }
+
+ /**
+ * @deprecated this method is to be used for pre 1.8 jdk.
+ */
+ @Deprecated
+ private void createThreadPool() {
+ ExecutorService executorService;
+ final ThreadGroup threadGroup =
+ LoggingThreadGroup.createThreadGroup("Socket asyncClose", logger);
+ ThreadFactory threadFactory = new ThreadFactory() {
+ public Thread newThread(final Runnable command) {
+ Thread thread = new Thread(threadGroup, command);
+ thread.setDaemon(true);
+ return thread;
+ }
+ };
+
+ executorService = new ThreadPoolExecutor(asyncClosePoolMaxThreads,
asyncClosePoolMaxThreads,
+ asyncCloseWaitTime, asyncCloseWaitUnits, new
LinkedBlockingQueue<>(), threadFactory);
}
/**
* Call this method if you know all the resources in the closer for the
given address are no
* longer needed. Currently a thread pool is kept for each address and
if you know that an address
* no longer needs its pool then you should call this method.
*/
- public void releaseResourcesForAddress(String address) {
- synchronized (asyncCloseExecutors) {
- ThreadPoolExecutor pool = asyncCloseExecutors.get(address);
- if (pool != null) {
- pool.shutdown();
- asyncCloseExecutors.remove(address);
- }
- }
- }
- private boolean isClosed() {
- synchronized (asyncCloseExecutors) {
- return this.closed;
+ public void releaseResourcesForAddress(String address) {
+ ExecutorService executorService = asyncCloseExecutors.remove(address);
+ if (executorService != null) {
+ executorService.shutdown();
--- End diff --
Correct and agreed. When the cache closes, I imagine SocketCloser.close()
is called. Which means the closing of any socket post that would be done
in-line.
I don't know what the expected behavior is supposed to be, for the closing
of the sockets when that cache is closing. But you are correct that it could
mean that the closing of the socket could be holding up the shutting down of
the cache.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
