geode-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kohlmu-pivotal <...@git.apache.org>
Subject [GitHub] geode pull request #702: GEODE-3416: Reduce synchronization blockages in Soc...
Date Tue, 15 Aug 2017 04:12:54 GMT
Github user kohlmu-pivotal commented on a diff in the pull request:

    https://github.com/apache/geode/pull/702#discussion_r133113689
  
    --- 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 --
    
    given that remove is on the concurrent hashmap, one should only ever get into this once.
This should not suffer reentrancy problems.


---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---

Mime
View raw message