/**
 * @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 void init_() {
    /* GC threads that haven't been used for 10 seconds. */
    pool_ = new ThreadPoolExecutor(
      getNumberOfThreads(),
      getNumberOfThreads(),
      10,
      TimeUnit.SECONDS,
      new LinkedBlockingQueue()
    );
  }

  public ExecutorService getPool() {
    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));
  }
}

@@ -1611,23 +1611,78 @@
 lic 
-void init_(
+synchronized ExecutorService getPool() %7B%0A    if ( pool_ == null 
 ) %7B%0A
+  
     
@@ -1741,16 +1741,18 @@
  */%0A    
+  
 pool_ = 
@@ -1775,16 +1775,18 @@
 ecutor(%0A
+  
       ge
@@ -1807,24 +1807,26 @@
 ds(),%0A      
+  
 getNumberOfT
@@ -1831,24 +1831,26 @@
 fThreads(),%0A
+  
       10,%0A  
@@ -1853,16 +1853,18 @@
 ,%0A      
+  
 TimeUnit
@@ -1873,16 +1873,18 @@
 ECONDS,%0A
+  
       ne
@@ -1908,11 +1908,23 @@
 ueue
+%3CRunnable%3E
 ()%0A
+  
     
@@ -1932,47 +1932,12 @@
 ;%0A  
+  
 %7D%0A
-%0A  public ExecutorService getPool() %7B
 %0A   

Fix ThreadPoolAgency initialization.