/**
 * @license
 * Copyright 2017 The FOAM Authors. All Rights Reserved.
 * http://www.apache.org/licenses/LICENSE-2.0
 */

package foam.nanos.pool;

import foam.core.Agency;
import foam.core.ContextAgent;
import foam.core.X;
import foam.nanos.logger.Logger;
import foam.nanos.pm.PM;
import java.util.concurrent.*;

public class ThreadPoolAgency
  extends    AbstractFixedThreadPool
  implements Agency
{
  protected ExecutorService pool_          = null;
  protected Object          queuedLock_    = new Object();
  protected Object          executingLock_ = new Object();
  protected Object          executedLock_  = new Object();

  class ContextAgentRunnable
    implements Runnable
  {
    final X            x_;
    final ContextAgent agent_;

    public ContextAgentRunnable(X x, ContextAgent agent) {
      x_     = x;
      agent_ = agent;
    }

    public void run() {
      incrExecuting(1);

      Logger logger = (Logger) x_.get("logger");
      PM     pm     = new PM(this.getClass(), agent_.getClass().getName());

      try {
        agent_.execute(x_);
      } catch (Throwable t) {
        logger.error(this.getClass(), t.getMessage(), t);
      } finally {
        incrExecuting(-1);
        incrExecuted();
        pm.log(getX());
      }
    }
  }

  public ThreadPoolAgency() {
  }

  public void incrExecuting(int d) {
    synchronized ( executingLock_ ) {
      executing_ += d;
    }
  }

  public void incrExecuted() {
    synchronized ( executedLock_ ) {
      executed_++;
    }
  }

  public void incrQueued() {
    synchronized ( queuedLock_ ) {
      queued_++;
    }
  }

  public synchronized ExecutorService getPool() {
    if ( pool_ == null ) {
      /* GC threads that haven't been used for 10 seconds. */
      pool_ = new ThreadPoolExecutor(
        getNumberOfThreads(),
        getNumberOfThreads(),
        10,
        TimeUnit.SECONDS,
        new LinkedBlockingQueue<Runnable>()
      );
    }

    return pool_;
  }

  // TODO: reverse order of 2nd and 3rd args
  public void submit(X x, ContextAgent agent, String description) {
    incrQueued();
    getPool().submit(new ContextAgentRunnable(x, agent));
  }
}

@@ -410,24 +410,34 @@
 otected 
+ThreadPool
 Executor
 Service 
@@ -428,23 +428,16 @@
 Executor
-Service
  pool_  
@@ -476,24 +476,27 @@
 ct          
+   
 queuedLock_ 
@@ -538,24 +538,27 @@
 ct          
+   
 executingLoc
@@ -600,24 +600,27 @@
 ct          
+   
 executedLock
@@ -1306,16 +1306,272 @@
 ncy() %7B%0A
+    /* GC threads that haven't been used for 10 seconds. */%0A    pool_ = new ThreadPoolExecutor(%0A      0,%0A      getNumberOfThreads(),%0A      10,%0A      TimeUnit.SECONDS,%0A      new LinkedBlockingQueue%3CRunnable%3E()%0A    );%0A    pool_.allowCoreThreadTimeOut(true);%0A
   %7D%0A%0A  p
@@ -1875,29 +1875,16 @@
  public 
-synchronized 
 Executor
@@ -1907,293 +1907,8 @@
 ) %7B%0A
-    if ( pool_ == null ) %7B%0A      /* GC threads that haven't been used for 10 seconds. */%0A      pool_ = new ThreadPoolExecutor(%0A        getNumberOfThreads(),%0A        getNumberOfThreads(),%0A        10,%0A        TimeUnit.SECONDS,%0A        new LinkedBlockingQueue%3CRunnable%3E()%0A      );%0A    %7D%0A%0A
     

Improve ThreadPool.