Java Code Examples for java.util.concurrent.ThreadPoolExecutor

/** 
 * Creates a new resource loader. The supplied <tt>oldResource</tt> will not be refreshed but a new one will replace it in the cache. This should be safe with regards to the rest of velocity since the default cache implementation is backed by a LRUMap discarding resources at will.
 * @param oldResource The resource being refreshed. Null if loaded for the first time.
 * @param name The name of the resource to load.
 * @param type The type of the resource to load.
 * @param enc The encoding of the resource.
 */
private Loader(final Resource oldResource,final String name,final int type,final String enc){
  this.oldResource=oldResource;
  this.name=name;
  this.type=type;
  this.enc=enc;
  key=type + name;
  if (LOG.isDebugEnabled() && EXECUTOR instanceof ThreadPoolExecutor) {
    final ThreadPoolExecutor tpe=(ThreadPoolExecutor)EXECUTOR;
    LOG.debug(DEBUG_POOL_COUNT + tpe.getActiveCount() + '/'+ tpe.getPoolSize());
  }
}
 

public ExtendedActivityMonitor(Instrumentation instrumentation,IntentFilter filter){
  activityInstanceMonitor=instrumentation.addMonitor(filter,null,false);
  this.activityMonitorThread=new Thread(){
    @Override public void run(){
      ThreadPoolExecutor executor=new ThreadPoolExecutor(2,5,5,TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(5));
      while (true) {
        Activity activity=activityInstanceMonitor.waitForActivityWithTimeout(CHECK_FOR_INTERRUPT_CYCLE_DURATION);
        long startTime=System.currentTimeMillis();
        if (activity != null) {
          int activitiesCount=startedActivities.size();
          if (activitiesCount > 0 && startedActivities.get(activitiesCount - 1).getActivity() == activity) {
            continue;
          }
synchronized (startedActivities) {
            startedActivities.add(new StartedActivity(activity,startTime));
          }
          Log.i(getClass().getSimpleName(),"Activity start: " + activity.getClass().getName());
          executor.submit(new ActivityStartListenerUpdater(activity));
        }
        if (interrupted() || stopped) {
          executor.shutdown();
          return;
        }
      }
    }
  }
;
}
 

/** 
 * Specify whether core threads are allowed to time out.  If the service is already started, the change will take effect immediately.
 * @param allowCoreTimeout {@code true} if core threads are allowed to time out
 */
public synchronized void setAllowCoreTimeout(final boolean allowCoreTimeout){
  this.allowCoreTimeout=allowCoreTimeout;
  final ThreadPoolExecutor realExecutor=this.realExecutor;
  if (realExecutor != null) {
    realExecutor.allowCoreThreadTimeOut(allowCoreTimeout);
  }
}
 

@SuppressWarnings("unchecked") public JobExecutorManager(FlowManager allKnownFlows,JobManager jobManager,Mailman mailman,String jobSuccessEmail,String jobFailureEmail,int maxThreads){
  this.jobManager=jobManager;
  this.mailman=mailman;
  this.jobSuccessEmail=jobSuccessEmail;
  this.jobFailureEmail=jobFailureEmail;
  this.allKnownFlows=allKnownFlows;
  Multimap<String,JobExecution> typedMultiMap=HashMultimap.create();
  this.completed=Multimaps.synchronizedMultimap(typedMultiMap);
  this.executing=new ConcurrentHashMap<String,ExecutingJobAndInstance>();
  this.executor=new ThreadPoolExecutor(0,maxThreads,10,TimeUnit.SECONDS,new LinkedBlockingQueue(),new ExecutorThreadFactory());
}
 

public static ThreadPoolExecutor coreThreadsTimoutExecutor(final int numberOfThreads,final int threadPriority,final String threadNamePrefix){
  final ThreadFactory factory=new ThreadFactoryBuilder().setPriority(threadPriority).setNameFormat(threadNamePrefix + "%d").setDaemon(true).build();
  final ThreadPoolExecutor pool=new ThreadPoolExecutor(numberOfThreads,numberOfThreads,100L,MILLISECONDS,new LinkedBlockingQueue<Runnable>(),factory);
  pool.allowCoreThreadTimeOut(true);
  return pool;
}
 

protected ExecutorService initializeExecutor(ThreadFactory threadFactory,RejectedExecutionHandler rejectedExecutionHandler){
  BlockingQueue<Runnable> queue=createQueue(this.queueCapacity);
  ThreadPoolExecutor executor=new ThreadPoolExecutor(this.corePoolSize,this.maxPoolSize,this.keepAliveSeconds,TimeUnit.SECONDS,queue,threadFactory,rejectedExecutionHandler);
  if (this.allowCoreThreadTimeOut) {
    executor.allowCoreThreadTimeOut(true);
  }
  this.exposedExecutor=(this.exposeUnconfigurableExecutor ? Executors.unconfigurableExecutorService(executor) : executor);
  return executor;
}
 

public void setMaxThreadCount(int maxThreadPoolCount){
  if (!(executor instanceof ThreadPoolExecutor)) {
    throw new IllegalStateException("cannot set max thread count for " + executor.getClass());
  }
  ThreadPoolExecutor threadPoolExecutor=(ThreadPoolExecutor)executor;
  threadPoolExecutor.setCorePoolSize(maxThreadPoolCount);
  threadPoolExecutor.setMaximumPoolSize(2 * maxThreadPoolCount);
}
 

private Executor getExecutor(int threads){
  if (threads <= 1) {
    return new Executor(){
      public void execute(      Runnable command){
        command.run();
      }
    }
;
  }
 else {
    return new ThreadPoolExecutor(threads,threads,3,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>());
  }
}
 

@SuppressWarnings("unchecked") public TaskControl(Comparator<PrioritizedTask> activeComparator,int maxThreads,ThreadFactory threadFactory,Log log){
  this.log=log;
  ApplicationIllegalArgumentException.notNull(activeComparator,"activeComparator");
  this.eligibleTasks=new PriorityBlockingQueue<PrioritizedTask>(20,activeComparator);
  this.stateChangeNotificator=new ReentrantLock();
  this.newTasks=this.stateChangeNotificator.newCondition();
  this.runningTasks=new AtomicInteger(0);
  this.threadFactory=threadFactory;
  int keepAliveTime=10;
  int corePoolSize=1;
  this.executor=new ThreadPoolExecutor(corePoolSize,Math.max(corePoolSize,maxThreads),keepAliveTime,MICROSECONDS,(BlockingQueue)this.eligibleTasks,threadFactory);
  this.stayActive=true;
}
 

@Inject public AsynchronousUpdateWorker(AggregatorConfig config){
  this.asyncDispatchQueue=new ArrayBlockingQueue<Runnable>(config.getAsyncUpdateWorkerBufferSize());
  executor=new ThreadPoolExecutor(config.getAsyncUpdateWorkerNumThreads(),config.getAsyncUpdateWorkerNumThreads(),60L,TimeUnit.SECONDS,this.asyncDispatchQueue,new NamedThreadFactory(getClass().getSimpleName()),new RejectedExecutionHandler(){
    @Override public void rejectedExecution(    Runnable r,    ThreadPoolExecutor executor){
      log.warn("AsynchronousUpdateWorker queue full, discarding task");
      discardedTasksDueToOverflow.getAndIncrement();
    }
  }
);
}
 

static public void main(String[] args) throws Exception {
  ThreadPoolExecutor executor=(ThreadPoolExecutor)Executors.newCachedThreadPool();
  ScheduledThreadPoolExecutor monitorExecutor=(ScheduledThreadPoolExecutor)Executors.newScheduledThreadPool(1,new ThreadFactory(){
    private AtomicInteger sequence=new AtomicInteger(0);
    @Override public Thread newThread(    Runnable r){
      Thread t=new Thread(r);
      t.setName("SmppServerSessionWindowMonitorPool-" + sequence.getAndIncrement());
      return t;
    }
  }
);
  SmppServerConfiguration configuration=new SmppServerConfiguration();
  configuration.setPort(2776);
  configuration.setMaxConnectionSize(10);
  configuration.setNonBlockingSocketsEnabled(true);
  configuration.setDefaultRequestExpiryTimeout(30000);
  configuration.setDefaultWindowMonitorInterval(15000);
  configuration.setDefaultWindowSize(5);
  configuration.setDefaultWindowWaitTimeout(configuration.getDefaultRequestExpiryTimeout());
  configuration.setDefaultSessionCountersEnabled(true);
  configuration.setJmxEnabled(true);
  DefaultSmppServer smppServer=new DefaultSmppServer(configuration,new DefaultSmppServerHandler(),executor,monitorExecutor);
  logger.info("Starting SMPP server...");
  smppServer.start();
  logger.info("SMPP server started");
  System.out.println("Press any key to stop server");
  System.in.read();
  logger.info("Stopping SMPP server...");
  smppServer.stop();
  logger.info("SMPP server stopped");
  logger.info("Server counters: {}",smppServer.getCounters());
}
 

public BatchWriter(EmbeddedSolrServer solr,int batchSize,TaskID tid,int writerThreads,int queueSize){
  this.solr=solr;
  this.writerThreads=writerThreads;
  this.queueSize=queueSize;
  taskId=tid;
  batchPool=new ThreadPoolExecutor(writerThreads,writerThreads,5,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(queueSize),new ThreadPoolExecutor.CallerRunsPolicy());
  this.batchToWrite=new ArrayList<SolrInputDocument>(batchSize);
}
 

private ThreadPoolManager(){
  _effectsScheduledThreadPool=new ScheduledThreadPoolExecutor(Config.THREAD_P_EFFECTS,new PriorityThreadFactory("EffectsSTPool",Thread.MIN_PRIORITY));
  _generalScheduledThreadPool=new ScheduledThreadPoolExecutor(Config.THREAD_P_GENERAL,new PriorityThreadFactory("GerenalSTPool",Thread.NORM_PRIORITY));
  _ioPacketsThreadPool=new ThreadPoolExecutor(2,Integer.MAX_VALUE,5L,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(),new PriorityThreadFactory("I/O Packet Pool",Thread.NORM_PRIORITY + 1));
  _generalPacketsThreadPool=new ThreadPoolExecutor(4,6,15L,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(),new PriorityThreadFactory("Normal Packet Pool",Thread.NORM_PRIORITY + 1));
  _generalThreadPool=new ThreadPoolExecutor(2,4,5L,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(),new PriorityThreadFactory("General Pool",Thread.NORM_PRIORITY));
  _aiThreadPool=new ThreadPoolExecutor(1,Config.AI_MAX_THREAD,10L,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>());
  _aiScheduledThreadPool=new ScheduledThreadPoolExecutor(Config.AI_MAX_THREAD,new PriorityThreadFactory("AISTPool",Thread.NORM_PRIORITY));
}
 

/** 
 * @param args
 */
public static void main(String[] args){
  Skills.initSkill(Const.PRODUCTION_EFFICIENCY_SKILL,(short)5);
  final ArrayBlockingQueue<Runnable> queue=new ArrayBlockingQueue<Runnable>(1000000);
  ThreadPoolExecutor executor=new ThreadPoolExecutor(20,20,5,TimeUnit.SECONDS,queue);
  try {
    for (    BlueprintDTO blueprint : blueprintDAO.getAllBlueprints()) {
      executor.execute(new PriceCalculatorRunnable(blueprint));
    }
  }
 catch (  SQLException e) {
    LOGGER.error(e.getMessage(),e);
  }
catch (  PopulatorException e) {
    LOGGER.error(e.getMessage(),e);
  }
  executor.shutdown();
}
 

public MultiThreadedObjectPipe(int numThreads,int queueSize,ObjectPipe delegate){
  this(new ThreadPoolExecutor(numThreads,numThreads,60L,TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(queueSize){
    public boolean offer(    Runnable runnable){
      try {
        return super.offer(runnable,Long.MAX_VALUE,TimeUnit.SECONDS);
      }
 catch (      InterruptedException e) {
        Thread.currentThread().interrupt();
        return false;
      }
    }
  }
),delegate);
}
 

public void sendToGroup(final String group,final RequestCommand request,final SingleRequestCallBackListener listener,final long time,final TimeUnit timeunut) throws NotifyRemotingException {
  if (group == null) {
    throw new NotifyRemotingException("Null group");
  }
  if (request == null) {
    throw new NotifyRemotingException("Null command");
  }
  if (listener == null) {
    throw new NotifyRemotingException("Null listener");
  }
  if (timeunut == null) {
    throw new NotifyRemotingException("Null TimeUnit");
  }
  final Connection conn=this.selectConnectionForGroup(group,this.connectionSelector,request);
  if (conn != null) {
    conn.send(request,listener,time,timeunut);
  }
 else {
    if (listener != null) {
      final ThreadPoolExecutor executor=listener.getExecutor();
      if (executor != null) {
        executor.execute(new Runnable(){
          public void run(){
            listener.onResponse(BaseRemotingController.this.createNoConnectionResponseCommand(request.getRequestHeader()),null);
          }
        }
);
      }
 else {
        listener.onResponse(this.createNoConnectionResponseCommand(request.getRequestHeader()),null);
      }
    }
  }
}
 

/** 
 * Shuts down the executor(s). Gracefully waits for  {@link #WAIT_FOR_JOBS_SHUTDOWN_TIMEOUT} seconds for all jobs to finishbefore forcefully shutting them down.
 */
public static void shutdown(){
  if (instance != null && instance.executor != null) {
    ThreadPoolExecutor pool=instance.executor;
    instance.executor=null;
    pool.shutdown();
    try {
      if (!pool.awaitTermination(WAIT_FOR_JOBS_SHUTDOWN_TIMEOUT,TimeUnit.SECONDS)) {
        pool.shutdownNow();
        if (!pool.awaitTermination(WAIT_FOR_JOBS_SHUTDOWN_TIMEOUT,TimeUnit.SECONDS)) {
          logger.error("Pool did not terminate");
        }
      }
    }
 catch (    InterruptedException ie) {
      pool.shutdownNow();
      Thread.currentThread().interrupt();
    }
  }
}
 

private void init(Configuration conf,ExecutorService es,int maxThreads) throws IOException {
  Validate.notNull(conf);
  this.conf=conf;
  if (maxThreads <= 0)   maxThreads=DEFAULT_MAX_THREADS;
 else   if (maxThreads < 3)   maxThreads=3;
  if (es == null) {
    ThreadPoolExecutor tpe=new ThreadPoolExecutor(3,maxThreads,5,TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(DEFAULT_QUEUE_SIZE));
    tpe.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
    closeables.addFirst(new IOUtil.ExecutorServiceCloseable(tpe,30));
    tpe.setThreadFactory(new ThreadFactory(){
      @Override public Thread newThread(      Runnable r){
        Thread t=Executors.defaultThreadFactory().newThread(r);
        t.setPriority(Thread.NORM_PRIORITY + 1);
        return t;
      }
    }
);
    tpe.prestartAllCoreThreads();
    es=tpe;
  }
  Validate.notNull(es);
  this.es=es;
  tpool=new HTablePool(conf,400);
  closeables.addFirst(tpool);
}
 

public AsyncTaskExecutor(){
  queue=new DelayQueue();
  executor=new ThreadPoolExecutor(10,500,5L,TimeUnit.SECONDS,queue,new ThreadFactoryBuilder().setDaemon(true).setNameFormat("AsyncTaskExecutor-" + instanceCounter.incrementAndGet() + "-%d").build()){
    protected <T>RunnableFuture<T> newTaskFor(    Runnable runnable,    T value){
      if (runnable instanceof DelayedRunnable) {
        return (FutureTask<T>)runnable;
      }
      return new FutureTask<T>(runnable,value);
    }
  }
;
}
 

/** 
 * Initializes threading parameters for the HTTPClient's  {@link ThreadSafeClientConnManager}, and this class' ThreadPoolExecutor. 
 */
protected void initThreadPools(final int poolSize,final ExecutorService threadPool){
  if (poolSize < 1)   throw new IllegalArgumentException("poolSize may not be < 1");
  HttpParams params=client != null ? client.getParams() : new BasicHttpParams();
  ConnManagerParams.setMaxTotalConnections(params,poolSize);
  ConnManagerParams.setMaxConnectionsPerRoute(params,new ConnPerRouteBean(poolSize));
  HttpProtocolParams.setVersion(params,HttpVersion.HTTP_1_1);
  SchemeRegistry schemeRegistry=new SchemeRegistry();
  schemeRegistry.register(new Scheme("http",PlainSocketFactory.getSocketFactory(),80));
  schemeRegistry.register(new Scheme("https",SSLSocketFactory.getSocketFactory(),443));
  ClientConnectionManager cm=new ThreadSafeClientConnManager(params,schemeRegistry);
  super.client=new DefaultHttpClient(cm,params);
  this.threadPool=threadPool != null ? threadPool : new ThreadPoolExecutor(poolSize,poolSize,120,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>());
}
 

@Inject private void init(){
  ThreadFactory storageThreadFactory=new ThreadFactoryBuilder().setNameFormat("Checkpointing-storage-%d").setUncaughtExceptionHandler(new UncaughtExceptionLogger("storage")).build();
  storageThreadPool=new ThreadPoolExecutor(1,storageMaxThreads,storageThreadKeepAliveSeconds,TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(storageMaxOutstandingRequests),storageThreadFactory,new ThreadPoolExecutor.CallerRunsPolicy());
  storageThreadPool.allowCoreThreadTimeOut(true);
  ThreadFactory serializationThreadFactory=new ThreadFactoryBuilder().setNameFormat("Checkpointing-serialization-%d").setUncaughtExceptionHandler(new UncaughtExceptionLogger("serialization")).build();
  serializationThreadPool=new ThreadPoolExecutor(1,serializationMaxThreads,serializationThreadKeepAliveSeconds,TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(serializationMaxOutstandingRequests),serializationThreadFactory,new ThreadPoolExecutor.AbortPolicy());
  serializationThreadPool.allowCoreThreadTimeOut(true);
  ThreadFactory fetchingThreadFactory=new ThreadFactoryBuilder().setNameFormat("Checkpointing-fetching-%d").setUncaughtExceptionHandler(new UncaughtExceptionLogger("fetching")).build();
  fetchingThreadPool=new ThreadPoolExecutor(0,fetchingMaxThreads,fetchingThreadKeepAliveSeconds,TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(fetchingQueueSize),fetchingThreadFactory);
  fetchingThreadPool.allowCoreThreadTimeOut(true);
}
 

public synchronized ThreadPoolExecutor startExecutor(){
  if (_executor == null) {
    LinkedBlockingQueue<Runnable> queue=new LinkedBlockingQueue<Runnable>(_maxExecQueueCapacity);
    _executor=new ThreadPoolExecutor(_maxThreads,_maxThreads,30,TimeUnit.SECONDS,queue);
    _executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
  }
  return _executor;
}
 

/** 
 * calcul du gradient en chaque beta 
 */
private double[] computeGrad(GeneralizedDoubleGaussL2 kernel){
  debug.print(3,"++++++ g : ");
  final double grad[]=new double[dim];
  int nbcpu=Runtime.getRuntime().availableProcessors();
  ThreadPoolExecutor threadPool=new ThreadPoolExecutor(nbcpu,nbcpu,10,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>());
  Queue<Future<?>> futures=new LinkedList<Future<?>>();
class GradRunnable implements Runnable {
    GeneralizedDoubleGaussL2 kernel;
    int i;
    public GradRunnable(    GeneralizedDoubleGaussL2 kernel,    int i){
      this.kernel=kernel;
      this.i=i;
    }
    public void run(){
      double[][] matrix=kernel.distanceMatrixUnthreaded(listOfExamples,i);
      double sum=0;
      for (int x=0; x < matrix.length; x++)       if (lambda_matrix[x] != null)       for (int y=0; y < matrix.length; y++)       sum+=matrix[x][y] * lambda_matrix[x][y];
      grad[i]+=0.5 * sum;
    }
  }
  for (int i=0; i < grad.length; i++) {
    Runnable r=new GradRunnable(kernel,i);
    futures.add(threadPool.submit(r));
  }
  while (!futures.isEmpty()) {
    try {
      futures.remove().get();
    }
 catch (    Exception e) {
      System.err.println("error with grad :");
      e.printStackTrace();
    }
  }
  threadPool.shutdownNow();
  for (int i=0; i < grad.length; i++)   if (Math.abs(grad[i]) < num_cleaning)   grad[i]=0.0;
  debug.println(3,Arrays.toString(grad));
  return grad;
}
 

/** 
 * @see org.jmangos.commons.service.Service#start()
 */
@PostConstruct @Override public void start(){
  final int scheduledPoolSize=ThreadPoolConfig.GENERAL_POOL;
  this.scheduledPool=new ScheduledThreadPoolExecutor(scheduledPoolSize);
  this.scheduledPool.prestartAllCoreThreads();
  final int instantPoolSize=ThreadPoolConfig.GENERAL_POOL;
  this.instantPool=new ThreadPoolExecutor(instantPoolSize,instantPoolSize,0,TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(100000));
  this.instantPool.prestartAllCoreThreads();
}
 

@Test public void testReadPerformance() throws Throwable {
  int size=10;
  ExecutorService es=new ThreadPoolExecutor(size,Integer.MAX_VALUE,30,TimeUnit.SECONDS,new SynchronousQueue());
  ((ThreadPoolExecutor)es).prestartAllCoreThreads();
  for (int i=0; i < size; i++)   es.execute(new WriteSpaceTask("PerformTask-" + i));
  ISOUtil.sleep(500);
  printAvg(t1,"Avg. write: ");
  for (int i=0; i < size; i++)   es.execute(new ReadSpaceTask("PerformTask-" + i));
  ISOUtil.sleep(500);
  printAvg(t2,"Avg. read : ");
  es.shutdown();
}
 

private static ExecutorService getBitmapExecutor(BitmapSource bitmapSource,BitmapSource.Type type){
  if (BitmapCache.getInstance().existOnDisk(bitmapSource.getAbsoluteFileName(type))) {
    if (sSingleThreadExecutor == null) {
      sSingleThreadExecutor=Executors.newSingleThreadExecutor(new BitmapLoaderThreadFactory("single thread"));
    }
    return sSingleThreadExecutor;
  }
 else {
    if (sBitmapExecutor == null) {
      sBitmapExecutor=new ThreadPoolExecutor(DEFAULT_CORE_POOL_SIZE,MAXIMUM_POOL_SIZE,KEEP_ALIVE,TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>(),new BitmapLoaderThreadFactory("multiple threads"));
    }
    return sBitmapExecutor;
  }
}
 

public RecoverStorageManager(final MetaClientConfig metaClientConfig,final SubscribeInfoManager subscribeInfoManager){
  super();
  this.threadPoolExecutor=new ThreadPoolExecutor(metaClientConfig.getRecoverThreadCount(),metaClientConfig.getRecoverThreadCount(),60,TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(100),new NamedThreadFactory("Recover-thread"),new ThreadPoolExecutor.CallerRunsPolicy());
  this.makeDataDir();
  this.subscribeInfoManager=subscribeInfoManager;
  this.loadStores();
}
 

/** 
 * Dispath a NOTIFY message to a new thread in the thread pool to send it. <br> If the thread pool is full, this method will block until it can get a spare thread to send the notify.
 * @param session
 * @param id
 * @return
 */
protected void sendIt(NotifyRequest nr){
  try {
    while (true) {
      try {
        _executor.execute(nr);
        break;
      }
 catch (      RejectedExecutionException e) {
        if (_executor instanceof ThreadPoolExecutor) {
          ThreadPoolExecutor exec=(ThreadPoolExecutor)_executor;
          LOG.warn("MemoryNotifySender thread pool is full. Thread num = " + exec.getPoolSize());
        }
 else {
          LOG.warn("MemoryNotifySender thread pool is full");
        }
        try {
          Thread.sleep(500);
        }
 catch (        InterruptedException e1) {
        }
      }
    }
  }
 catch (  Throwable t) {
    LOG.error("MemoryNotifyDispatcher Error",t);
  }
 finally {
    ;
  }
}
 

/** 
 * Create a PeerGroup
 */
public PeerGroup(BlockStore blockStore,NetworkParameters params,BlockChain chain){
  this.blockStore=blockStore;
  this.params=params;
  this.chain=chain;
  inactives=new LinkedBlockingQueue<PeerAddress>();
  peers=Collections.synchronizedSet(new HashSet<Peer>());
  peerEventListeners=Collections.synchronizedSet(new HashSet<PeerEventListener>());
  pendingTransactionListeners=Collections.synchronizedList(new ArrayList<PendingTransactionListener>());
  peerPool=new ThreadPoolExecutor(CORE_THREADS,DEFAULT_CONNECTIONS,THREAD_KEEP_ALIVE_SECONDS,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(1),new PeerGroupThreadFactory());
}
 

public AsyncEventExecutor(int poolSize,int maxPoolSize,int keepAliveTime,int queueSize){
  queue=new LinkedBlockingQueue<Runnable>(queueSize);
  mono_queue=new LinkedBlockingQueue<Runnable>(queueSize);
  NamedThreadFactory threadFactory=new NamedThreadFactory("Nuxeo Async Events");
  executor=new ThreadPoolExecutor(poolSize,maxPoolSize,keepAliveTime,TimeUnit.SECONDS,queue,threadFactory);
  mono_executor=new ThreadPoolExecutor(1,1,keepAliveTime,TimeUnit.SECONDS,mono_queue,threadFactory);
}
 

/** 
 * Build the weight table, parallelized according to the number of processors 
 */
public void buildTable(){
  ArrayList<TransitStop> stopVertices;
  LOG.debug("Number of vertices: " + g.getVertices().size());
  stopVertices=new ArrayList<TransitStop>();
  for (  Vertex gv : g.getVertices())   if (gv instanceof TransitStop)   stopVertices.add((TransitStop)gv);
  int nStops=stopVertices.size();
  stopIndices=new IdentityHashMap<Vertex,Integer>(nStops);
  for (int i=0; i < nStops; i++)   stopIndices.put(stopVertices.get(i),i);
  LOG.debug("Number of stops: " + nStops);
  table=new float[nStops][nStops];
  for (  float[] row : table)   Arrays.fill(row,Float.POSITIVE_INFINITY);
  LOG.debug("Performing search at each transit stop.");
  int nThreads=Runtime.getRuntime().availableProcessors();
  LOG.debug("number of threads: " + nThreads);
  ArrayBlockingQueue<Runnable> taskQueue=new ArrayBlockingQueue<Runnable>(nStops);
  ThreadPoolExecutor threadPool=new ThreadPoolExecutor(nThreads,nThreads,10,TimeUnit.SECONDS,taskQueue);
  GenericObjectPool heapPool=new GenericObjectPool(new PoolableBinHeapFactory<State>(g.getVertices().size()),nThreads);
  RoutingRequest options=new RoutingRequest();
  options.setWalkSpeed(maxWalkSpeed);
  final double MAX_WEIGHT=60 * 60 * options.walkReluctance;
  final double OPTIMISTIC_BOARD_COST=options.getBoardCostLowerBound();
  ArrayList<Callable<Void>> tasks=new ArrayList<Callable<Void>>();
  for (  TransitStop origin : stopVertices) {
    SPTComputer task=new SPTComputer(heapPool,options,MAX_WEIGHT,OPTIMISTIC_BOARD_COST,origin);
    tasks.add(task);
  }
  try {
    threadPool.invokeAll(tasks);
    threadPool.shutdown();
  }
 catch (  InterruptedException e) {
    throw new RuntimeException(e);
  }
  floyd();
}
 

public ConnectionBase(final ConnectionInformation connectionInformation){
  super();
  this.connectionInformation=connectionInformation;
  this.lookupExecutor=new ThreadPoolExecutor(0,1,1,TimeUnit.MINUTES,new LinkedBlockingQueue<Runnable>(),new NamedThreadFactory("ConnectionBaseExecutor/" + connectionInformation.toMaskedString()));
  this.messenger=new Messenger(getMessageTimeout());
  this.pingService=new PingService(this.messenger);
  this.connector=createConnector();
}
 

public StorageImpl() throws IOException {
  this.taskQueue=new LinkedBlockingQueue<Runnable>();
  this.executorService=new ThreadPoolExecutor(1,1,0L,TimeUnit.MILLISECONDS,this.taskQueue,new NamedThreadFactory(StorageImpl.class.getName()));
  this.rootFolder=new File(System.getProperty("org.openscada.ds.storage.file.root",System.getProperty("user.home") + File.separator + ".openscadaDS"));
  if (!this.rootFolder.exists()) {
    this.rootFolder.mkdirs();
  }
  if (!this.rootFolder.exists() || !this.rootFolder.isDirectory()) {
    throw new IOException(String.format("Unable to use directory: %s",this.rootFolder));
  }
}
 

public BatchWriter(EmbeddedSolrServer solr,int batchSize,TaskID tid,int writerThreads,int queueSize){
  this.solr=solr;
  this.writerThreads=writerThreads;
  this.queueSize=queueSize;
  taskId=tid;
  batchPool=new ThreadPoolExecutor(writerThreads,writerThreads,5,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(queueSize),new ThreadPoolExecutor.CallerRunsPolicy());
  this.batchToWrite=new ArrayList<SolrInputDocument>(batchSize);
}
 

@Override public void performOperation() throws OperationFailedException {
  Logger logger=Logger.getLogger(Pipeline.primaryLoggerName);
  ThreadPoolExecutor threadPool=(ThreadPoolExecutor)Executors.newFixedThreadPool(getPreferredThreadCount());
  Date now=new Date();
  long beginMillis=System.currentTimeMillis();
  logger.info("[" + now + "] ParallelOperator is launching "+ operators.size()+ " simultaneous jobs");
  this.getPipelineOwner().fireMessage("ParallelOperator is launching " + operators.size() + " jobs");
  List<OpWrapper> wraps=new ArrayList<OpWrapper>();
  for (  Operator op : operators) {
    OpWrapper opw=new OpWrapper(op);
    wraps.add(opw);
    threadPool.submit(opw);
  }
  threadPool.shutdown();
  try {
    System.out.println("Awaiting termingation of thread pool");
    threadPool.awaitTermination(7,TimeUnit.DAYS);
    System.out.println("Threadpool has terminated");
  }
 catch (  InterruptedException e1) {
    throw new OperationFailedException("Parallel Operator " + this.getObjectLabel() + " was interrupted during parallel execution",this);
  }
  now=new Date();
  long endMillis=System.currentTimeMillis();
  long elapsedMillis=endMillis - beginMillis;
  logger.info("[ " + now + "] Parallel operator: "+ getObjectLabel()+ " has completed. Time taken = "+ elapsedMillis+ " ms ( "+ ElapsedTimeFormatter.getElapsedTime(beginMillis,endMillis)+ " )");
}
 

BackgroundLoaderManager(Context context){
  mPendingTaskUris=new HashSet<Uri>();
  mCallbacks=new HashMap<Uri,Set<ItemLoadedCallback>>();
  final LinkedBlockingQueue<Runnable> queue=new LinkedBlockingQueue<Runnable>();
  final int poolSize=MAX_THREADS;
  mExecutor=new ThreadPoolExecutor(poolSize,poolSize,5,TimeUnit.SECONDS,queue,new BackgroundLoaderThreadFactory(getTag()));
  mCallbackHandler=new Handler();
}
 

/** 
 * @hide
 */
public static ExecutorService newConfiguredThreadPool(){
  int corePoolSize=0;
  int maximumPoolSize=Configuration.ASYNC_THREAD_COUNT;
  long keepAliveTime=60L;
  TimeUnit unit=TimeUnit.SECONDS;
  BlockingQueue<Runnable> workQueue=new LinkedBlockingQueue<Runnable>();
  RejectedExecutionHandler handler=new ThreadPoolExecutor.CallerRunsPolicy();
  return new ThreadPoolExecutor(corePoolSize,maximumPoolSize,keepAliveTime,unit,workQueue,handler);
}
 

private void setupResource(){
  mCacher=new Cache(getApplicationContext());
  mThreadPoolExecutor=new ThreadPoolExecutor(10,20,10,TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(100));
  mAPIClient=new RCAPIClient(getApplicationContext(),new JSONParser(),mThreadPoolExecutor,mCacher);
  mImgLoader=new BitmapAsyncLoader(new AsyncLoaderEngine(getApplicationContext(),mThreadPoolExecutor,mCacher));
}
 

public boolean isAllowCoreThreadTimeOut(){
  if (this.internalExecutor != null) {
    ThreadPoolExecutor executor=this.internalExecutor.getThreadPoolExecutor();
    try {
      Method m=ThreadPoolExecutor.class.getMethod("allowsCoreThreadTimeOut",null);
      try {
        return (Boolean)m.invoke(executor,null);
      }
 catch (      Exception ex) {
      }
    }
 catch (    NoSuchMethodException ex) {
    }
  }
  return false;
}
 

public synchronized static ThreadPoolExecutor getExecutorService(){
  try {
    if (threadPool == null) {
      threadPool=new ThreadPoolExecutor(1,1,0,TimeUnit.SECONDS,(new ArrayBlockingQueue(100)));
    }
    return threadPool;
  }
 catch (  Exception jdo) {
    jdo.printStackTrace();
    System.out.println("I couldn't instantiate the mailThreadExecutorService.");
    return null;
  }
}
 

public Networker(NetApp napp,Context ctx,ScrobblesDatabase db){
  settings=new AppSettings(ctx);
  mNetApp=napp;
  mCtx=ctx;
  mDb=db;
  mComparator=new NetRunnableComparator();
  mExecutor=new ThreadPoolExecutor(1,1,2,TimeUnit.SECONDS,new PriorityBlockingQueue<Runnable>(1,mComparator));
  mSleeper=new Sleeper(mNetApp,ctx,this);
  mNetworkWaiter=new NetworkWaiter(mNetApp,ctx,this);
  hInfo=null;
}
 

/** 
 * @return the threadPool
 */
public ThreadPoolExecutor getDeliveryExecuter(){
  if (!this.started || this.suspended) {
    return null;
  }
  if (this.deliveryExecuter == null) {
    this.deliveryExecuter=new ThreadPoolExecutor(this.minThreads,this.maxThreads,DefaultDeliveryManager.DEFAULT_KEEPALIVE_TIME,TimeUnit.MILLISECONDS,this.workQueue);
  }
  return this.deliveryExecuter;
}
 

private Executor getExecutor(int threads){
  if (threads <= 1) {
    return new Executor(){
      public void execute(      Runnable command){
        command.run();
      }
    }
;
  }
 else {
    return new ThreadPoolExecutor(threads,threads,3,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>());
  }
}
 

private void runSubmitTest(Object result) throws InterruptedException, ExecutionException, TimeoutException {
  ExecutorService executor=new ThreadPoolExecutor(5,5,5L,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(5));
  SignallingRunnable runner=new SignallingRunnable("runSubmitTest(result=" + result + ")");
  Future<?> future=(result == null) ? executor.submit(runner) : executor.submit(runner,result);
  assertNotNull("No future instance returned",future);
  assertLastExecutionOperation(runner);
  assertCurrentThreadExecution();
  if (result != null) {
    Object actual=future.get(5L,TimeUnit.SECONDS);
    assertEquals("Mismatched future result",result,actual);
  }
}
 

/** 
 * Creates a new custom thread pool
 * @param pattern                  pattern of the thread name
 * @param name                     ${name} in the pattern name
 * @param corePoolSize             the core size
 * @param maxPoolSize              the maximum pool size
 * @param keepAliveTime            keep alive time
 * @param timeUnit                 keep alive time unit
 * @param maxQueueSize             the maximum number of tasks in the queue, use <tt>Integer.MAX_VALUE</tt> or <tt>-1</tt> to indicate unbounded
 * @param rejectedExecutionHandler the handler for tasks which cannot be executed by the thread pool.If <tt>null</tt> is provided then  {@link java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy CallerRunsPolicy} is used.
 * @param daemon                   whether the threads is daemon or not
 * @return the created pool
 * @throws IllegalArgumentException if parameters is not valid
 */
public static ExecutorService newThreadPool(final String pattern,final String name,int corePoolSize,int maxPoolSize,long keepAliveTime,TimeUnit timeUnit,int maxQueueSize,RejectedExecutionHandler rejectedExecutionHandler,final boolean daemon){
  if (maxPoolSize < corePoolSize) {
    throw new IllegalArgumentException("MaxPoolSize must be >= corePoolSize, was " + maxPoolSize + " >= "+ corePoolSize);
  }
  BlockingQueue<Runnable> queue;
  if (corePoolSize == 0 && maxQueueSize <= 0) {
    queue=new SynchronousQueue<Runnable>();
    corePoolSize=1;
    maxPoolSize=1;
  }
 else   if (maxQueueSize <= 0) {
    queue=new LinkedBlockingQueue<Runnable>();
  }
 else {
    queue=new LinkedBlockingQueue<Runnable>(maxQueueSize);
  }
  ThreadPoolExecutor answer=new ThreadPoolExecutor(corePoolSize,maxPoolSize,keepAliveTime,timeUnit,queue);
  answer.setThreadFactory(new ThreadFactory(){
    public Thread newThread(    Runnable r){
      Thread answer=new Thread(r,getThreadName(pattern,name));
      answer.setDaemon(daemon);
      return answer;
    }
  }
);
  if (rejectedExecutionHandler == null) {
    rejectedExecutionHandler=new ThreadPoolExecutor.CallerRunsPolicy();
  }
  answer.setRejectedExecutionHandler(rejectedExecutionHandler);
  return answer;
}
 

@Test public void main() throws InterruptedException {
  BlockingQueue<Runnable> queue=new LinkedBlockingQueue<Runnable>();
  ThreadPoolExecutor executor=new ThreadPoolExecutor(SIZE,SIZE,1000,TimeUnit.MILLISECONDS,queue);
  long time=System.currentTimeMillis();
  for (int i=0; i < SIZE; i++) {
    executor.submit(new Runnable(){
      final Tedis tedis=new Tedis(ip1);
      public void run(){
        for (int j=0; j <= 2000; j++) {
          tedis.set("foo".getBytes(),"bar".getBytes());
          if (j % 1000 == 0) {
            System.out.println("finished:" + j);
          }
        }
      }
    }
);
  }
  executor.shutdown();
  while (!executor.isTerminated()) {
  }
  System.out.println("qps:" + (SIZE * 2000) / ((System.currentTimeMillis() - time) / 1000));
}
 

public TThreadPoolServer(TProcessorFactory processorFactory,TServerTransport serverTransport,TTransportFactory inputTransportFactory,TTransportFactory outputTransportFactory,TProtocolFactory inputProtocolFactory,TProtocolFactory outputProtocolFactory,Options options){
  super(processorFactory,serverTransport,inputTransportFactory,outputTransportFactory,inputProtocolFactory,outputProtocolFactory);
  executorService_=null;
  SynchronousQueue<Runnable> executorQueue=new SynchronousQueue<Runnable>();
  executorService_=new ThreadPoolExecutor(options.minWorkerThreads,options.maxWorkerThreads,60,TimeUnit.SECONDS,executorQueue);
  options_=options;
}
 

/** 
 * Start accepting new connections in a background thread.
 */
public void start(){
  if (!this.socket.isBound()) {
    throw new IllegalStateException("Can't start ConnectionHandler " + "without a bound socket!");
  }
  this.stop=false;
  if (this.executor == null || this.executor.isShutdown()) {
    this.executor=new ThreadPoolExecutor(OUTBOUND_CONNECTIONS_POOL_SIZE,OUTBOUND_CONNECTIONS_POOL_SIZE,OUTBOUND_CONNECTIONS_THREAD_KEEP_ALIVE_SECS,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(),new ConnectorThreadFactory());
  }
  if (this.thread == null || !this.thread.isAlive()) {
    this.thread=new Thread(this);
    this.thread.setName("bt-serve");
    this.thread.start();
  }
}
 

/** 
 * Create the structure to enable submission by database
 * @throws WaarpDatabaseNoConnectionException
 * @throws WaarpDatabaseSqlException
 */
public InternalRunner() throws WaarpDatabaseNoConnectionException, WaarpDatabaseSqlException {
  if (DbConstant.admin.isConnected) {
    commander=new Commander(this,true);
  }
 else {
    commander=new CommanderNoDb(this,true);
  }
  scheduledExecutorService=Executors.newSingleThreadScheduledExecutor();
  isRunning=true;
  workQueue=new ArrayBlockingQueue<Runnable>(10);
  threadPoolExecutor=new ThreadPoolExecutor(10,Configuration.configuration.RUNNER_THREAD,1000,TimeUnit.MILLISECONDS,workQueue);
  scheduledFuture=scheduledExecutorService.scheduleWithFixedDelay(commander,Configuration.configuration.delayCommander,Configuration.configuration.delayCommander,TimeUnit.MILLISECONDS);
  networkTransaction=new NetworkTransaction();
}
 

/** 
 * @param args
 * @throws Exception
 */
public static void main(String[] args) throws Exception {
  String encoding=System.getProperty("file.encoding");
  if (!"UTF-8".equals(encoding)) {
    throw new UTFDataFormatException("system encoding " + encoding + "is not UTF-8");
  }
  logger.configureLogger(System.getProperties());
  int threads=Integer.parseInt(System.getProperty("THREADS","" + Runtime.getRuntime().availableProcessors()));
  int capacity=1000 * threads;
  BlockingQueue<Runnable> workQueue=new ArrayBlockingQueue<Runnable>(capacity);
  ThreadPoolExecutor threadPoolExecutor=new ThreadPoolExecutor(threads,threads,60,TimeUnit.SECONDS,workQueue);
  threadPoolExecutor.prestartAllCoreThreads();
  File file;
  FragmentZipFiles factory=new FragmentZipFiles();
  for (int i=0; i < args.length; i++) {
    file=new File(args[i]);
    FragmentTask task=factory.new FragmentTask(file);
    threadPoolExecutor.submit(task);
  }
  threadPoolExecutor.shutdown();
  while (!threadPoolExecutor.isTerminated()) {
    threadPoolExecutor.awaitTermination(5,TimeUnit.SECONDS);
  }
  logger.info("all files completed");
}
 

public void start(){
  ThreadFactory tf=new ThreadFactoryBuilder().setNameFormat("ContainerLauncher #%d").setDaemon(true).build();
  launcherPool=new ThreadPoolExecutor(INITIAL_POOL_SIZE,Integer.MAX_VALUE,1,TimeUnit.HOURS,new LinkedBlockingQueue<Runnable>(),tf);
  eventHandlingThread=new Thread(){
    @Override public void run(){
      ContainerLauncherEvent event=null;
      while (!Thread.currentThread().isInterrupted()) {
        try {
          event=eventQueue.take();
        }
 catch (        InterruptedException e) {
          LOG.error("Returning, interrupted : " + e);
          return;
        }
        int poolSize=launcherPool.getCorePoolSize();
        if (poolSize != limitOnPoolSize) {
          int numNodes=allNodes.size();
          int idealPoolSize=Math.min(limitOnPoolSize,numNodes);
          if (poolSize < idealPoolSize) {
            int newPoolSize=Math.min(limitOnPoolSize,idealPoolSize + INITIAL_POOL_SIZE);
            LOG.info("Setting ContainerLauncher pool size to " + newPoolSize + " as number-of-nodes to talk to is "+ numNodes);
            launcherPool.setCorePoolSize(newPoolSize);
          }
        }
        launcherPool.execute(createEventProcessor(event));
      }
    }
  }
;
  eventHandlingThread.setName("ContainerLauncher Event Handler");
  eventHandlingThread.start();
  super.start();
}
 

/** 
 * So there is this factory where all workers do is running and then relax at the pool, and where all clients must wait in a queue. It is a pretty fun work environment... until everyone gets garbage collected that is.
 */
private ExecutorService workers(){
  return new ThreadPoolExecutor(5,10,90,TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(100),new ThreadFactory(){
    @Override public Thread newThread(    final Runnable r){
      Thread worker=new Thread(r){
        @Override public void run(){
          super.run();
        }
      }
;
      worker.setUncaughtExceptionHandler(new UncaughtExceptionHandler(){
        @Override public void uncaughtException(        final Thread t,        final Throwable e){
          log.error("Uncaught exception from bagheera load worker.",e);
        }
      }
);
      return worker;
    }
  }
,new RejectedExecutionHandler(){
    @Override public void rejectedExecution(    Runnable task,    ThreadPoolExecutor executor){
      try {
        executor.getQueue().put(task);
      }
 catch (      InterruptedException e) {
        throw new RuntimeException(e);
      }
    }
  }
);
}
 

/** 
 * Prueba la efectividad de  {@link BlockingRejectedExecutionHandler} utilizando una cola bloqueante. 
 */
@Test(timeout=2000) public final void foo() throws InterruptedException {
  final ExecutorService s=new ThreadPoolExecutor(1,1,0,TimeUnit.MILLISECONDS,new SynchronousQueue<Runnable>(),new BlockingRejectedExecutionHandler());
  final AtomicInteger i=new AtomicInteger(0);
  final CountDownLatch latch=new CountDownLatch(1);
  final CountDownLatch end=new CountDownLatch(1);
  s.submit(new Runnable(){
    public void run(){
      try {
        latch.await();
        Thread.sleep(1000);
        i.incrementAndGet();
      }
 catch (      InterruptedException e) {
        throw new UnhandledException(e);
      }
    }
  }
);
  latch.countDown();
  s.submit(new Runnable(){
    public void run(){
      i.incrementAndGet();
      end.countDown();
    }
  }
);
  end.await();
  s.shutdown();
  while (!s.awaitTermination(500,TimeUnit.MILLISECONDS)) {
  }
  Assert.assertEquals(2,i.get());
}
 

private RunResult runSuitesForkPerTestSet(final Properties properties,final SurefireProperties effectiveSystemProperties,int forkCount) throws SurefireBooterForkException {
  ArrayList<Future<RunResult>> results=new ArrayList<Future<RunResult>>(500);
  ExecutorService executorService=new ThreadPoolExecutor(forkCount,forkCount,60,TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(500));
  try {
    RunResult globalResult=new RunResult(0,0,0,0);
    final Iterator suites=getSuitesIterator();
    while (suites.hasNext()) {
      final Object testSet=suites.next();
      final ForkClient forkClient=new ForkClient(fileReporterFactory,startupReportConfiguration.getTestVmSystemProperties());
      Callable<RunResult> pf=new Callable<RunResult>(){
        public RunResult call() throws Exception {
          return fork(testSet,new PropertiesWrapper(properties),forkClient,fileReporterFactory.getGlobalRunStatistics(),effectiveSystemProperties);
        }
      }
;
      results.add(executorService.submit(pf));
    }
    for (    Future<RunResult> result : results) {
      try {
        RunResult cur=result.get();
        if (cur != null) {
          globalResult=globalResult.aggregate(cur);
        }
 else {
          throw new SurefireBooterForkException("No results for " + result.toString());
        }
      }
 catch (      InterruptedException e) {
        throw new SurefireBooterForkException("Interrupted",e);
      }
catch (      ExecutionException e) {
        throw new SurefireBooterForkException("ExecutionException",e);
      }
    }
    return globalResult;
  }
  finally {
    closeExecutor(executorService);
  }
}
 

public MultiThreader(){
  int numThreads=Runtime.getRuntime().availableProcessors();
  BlockingQueue<Runnable> queue=new ArrayBlockingQueue<Runnable>(1);
  exec=new ThreadPoolExecutor(numThreads,numThreads,0,TimeUnit.SECONDS,queue,new ThreadFactory(){
    private final AtomicInteger seq=new AtomicInteger(1);
    @Override public Thread newThread(    Runnable runnable){
      return new Thread("MultiThreader-" + seq.getAndIncrement()){
        @Override public void run(){
          try {
            super.run();
          }
 catch (          Throwable t) {
            failure.compareAndSet(null,t);
          }
        }
      }
;
    }
  }
);
  failure=new AtomicReference<Throwable>();
}
 

@Override public ExecutorService initClusterTaskExecutor(RejectedExecutionHandler handler) throws PlatformException {
  BlockingQueue<Runnable> actionQueue=new ArrayBlockingQueue<Runnable>(getActionQueueSize());
  taskExec=new ThreadPoolExecutor(getActionMinThreads(),getActionMaxThreads(),getActionThreadTimeout(),TimeUnit.SECONDS,actionQueue,new ThreadFactory(){
    private final ThreadFactory factory=java.util.concurrent.Executors.defaultThreadFactory();
    private long id=0;
    @Override public Thread newThread(    Runnable r){
      Thread t=factory.newThread(r);
      t.setName("ODE-X Cluster Action Executor - " + ++id);
      t.setDaemon(true);
      return t;
    }
  }
,handler);
  taskExec.allowCoreThreadTimeOut(true);
  return taskExec;
}
 

public void start() throws Exception {
  initIOHandler();
  ByteBuffer.setUseDirectBuffers(!useHeapBuffers);
  if (useHeapBuffers)   ByteBuffer.setAllocator(new SimpleByteBufferAllocator());
  log.info("MRTMP Mina Transport Settings");
  log.info("IO Threads: " + ioThreads);
  log.info("Event Threads:" + " core: " + eventThreadsCore + "+1"+ " max: "+ eventThreadsMax+ "+1"+ " queue: "+ eventThreadsQueue+ " keepalive: "+ eventThreadsKeepalive);
  eventExecutor=new ThreadPoolExecutor(eventThreadsCore + 1,eventThreadsMax + 1,eventThreadsKeepalive,TimeUnit.SECONDS,threadQueue(eventThreadsQueue));
  ((ThreadPoolExecutor)eventExecutor).setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
  acceptor=new SocketAcceptor(ioThreads,Executors.newCachedThreadPool());
  acceptor.getFilterChain().addLast("threadPool",new ExecutorFilter(eventExecutor));
  SocketAcceptorConfig config=acceptor.getDefaultConfig();
  config.setThreadModel(ThreadModel.MANUAL);
  config.setReuseAddress(true);
  config.setBacklog(100);
  log.info("TCP No Delay: " + tcpNoDelay);
  log.info("Receive Buffer Size: " + receiveBufferSize);
  log.info("Send Buffer Size: " + sendBufferSize);
  SocketSessionConfig sessionConf=(SocketSessionConfig)config.getSessionConfig();
  sessionConf.setReuseAddress(true);
  sessionConf.setTcpNoDelay(tcpNoDelay);
  if (isLoggingTraffic) {
    log.info("Configuring traffic logging filter");
    IoFilter filter=new LoggingFilter();
    acceptor.getFilterChain().addFirst("LoggingFilter",filter);
  }
  SocketAddress socketAddress=(address == null) ? new InetSocketAddress(port) : new InetSocketAddress(address,port);
  acceptor.bind(socketAddress,ioHandler);
  log.info("MRTMP Mina Transport bound to " + socketAddress.toString());
}
 

/** 
 * <p> This init() method <b>must</b> be called by the dependency injection framework. It waits until all required dependencies are injected in SafeKeeper, and until the node count is accessible from the communication layer. </p>
 */
public void init(){
  try {
    getReadySignal().await();
  }
 catch (  InterruptedException e1) {
    e1.printStackTrace();
  }
  threadPool=new ThreadPoolExecutor(1,maxWriteThreads,writeThreadKeepAliveSeconds,TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(maxOutstandingWriteRequests));
  logger.debug("Started thread pool with maxWriteThreads=[" + maxWriteThreads + "], writeThreadKeepAliveSeconds=["+ writeThreadKeepAliveSeconds+ "], maxOutsandingWriteRequests=["+ maxOutstandingWriteRequests+ "]");
  int nodeCount=getLoopbackDispatcher().getEventEmitter().getNodeCount();
  while (nodeCount == 0) {
    try {
      Thread.sleep(500);
    }
 catch (    InterruptedException ignored) {
    }
    nodeCount=getLoopbackDispatcher().getEventEmitter().getNodeCount();
  }
  signalNodesAvailable.countDown();
}
 

/** 
 * Constructs an instance of the object pool, including handlers, requests and responses
 */
public ObjectPool(Map args) throws IOException {
  this.simulateModUniqueId=Option.SIMULATE_MOD_UNIQUE_ID.get(args);
  this.saveSessions=Option.USE_SAVED_SESSIONS.get(args);
  this.maxConcurrentRequests=Option.HANDLER_COUNT_MAX.get(args);
  this.maxIdleRequestHandlersInPool=Option.HANDLER_COUNT_MAX_IDLE.get(args);
  ExecutorService es=new ThreadPoolExecutor(maxIdleRequestHandlersInPool,Integer.MAX_VALUE,60L,TimeUnit.SECONDS,new SynchronousQueue<Runnable>(),new ThreadFactory(){
    private int threadIndex;
    public synchronized Thread newThread(    Runnable r){
      String threadName=Launcher.RESOURCES.getString("RequestHandlerThread.ThreadName","" + (++threadIndex));
      Thread thread=new Thread(r,threadName);
      thread.setDaemon(true);
      return thread;
    }
  }
);
  requestHandler=new BoundedExecutorService(es,maxConcurrentRequests);
  this.unusedRequestPool=new ArrayList();
  this.unusedResponsePool=new ArrayList();
  for (int n=0; n < START_REQUESTS_IN_POOL; n++) {
    this.unusedRequestPool.add(new WinstoneRequest());
  }
  for (int n=0; n < START_RESPONSES_IN_POOL; n++) {
    this.unusedResponsePool.add(new WinstoneResponse());
  }
}
 

/** 
 * @see java.lang.Runnable#run()
 */
public void run(){
  final ExecutorService handlers=new ThreadPoolExecutor(1,5,60L,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>());
  final ObjectName mbeanName=JMXHelper.register(new Agent(),"org.kjkoster.zapcat:type=Agent,port=" + port);
  try {
    serverSocket=new ServerSocket(port,0,address);
    while (!stopping) {
      final Socket accepted=serverSocket.accept();
      log.debug("accepted connection from " + accepted.getInetAddress().getHostAddress());
      if (acceptedByWhitelist(accepted.getInetAddress())) {
        handlers.execute(new QueryHandler(accepted));
      }
 else {
        log.warn("rejecting ip address " + accepted.getInetAddress().getHostAddress() + ", it is not on the whitelist");
      }
    }
  }
 catch (  IOException e) {
    if (!stopping) {
      log.error("caught exception, exiting",e);
    }
  }
 finally {
    try {
      if (serverSocket != null) {
        serverSocket.close();
        serverSocket=null;
      }
    }
 catch (    IOException e) {
    }
    try {
      handlers.shutdown();
      handlers.awaitTermination(10,TimeUnit.SECONDS);
    }
 catch (    InterruptedException e) {
    }
    JMXHelper.unregister(mbeanName);
  }
}
 

/** 
 * Converts the given ThreadPoolExecutor into an ExecutorService that exits when the application is complete.  It does so by using daemon threads and adding a shutdown hook to wait for their completion. <p>This is mainly for fixed thread pools. See  {@link java.util.concurrent.Executors#newFixedThreadPool(int)}.
 * @param executor the executor to modify to make sure it exits when theapplication is finished
 * @param terminationTimeout how long to wait for the executor tofinish before terminating the JVM
 * @param timeUnit unit of time for the time parameter
 * @return an unmodifiable version of the input which will not hang the JVM
 */
public static ExecutorService getExitingExecutorService(ThreadPoolExecutor executor,long terminationTimeout,TimeUnit timeUnit){
  executor.setThreadFactory(daemonThreadFactory(executor.getThreadFactory()));
  ExecutorService service=java.util.concurrent.Executors.unconfigurableExecutorService(executor);
  addDelayedShutdownHook(service,terminationTimeout,timeUnit);
  return service;
}
 

private synchronized ThreadPoolExecutor getPool(){
  if (pool == null) {
    pool=createPool();
    logger.info("Thread pool started.");
  }
  return pool;
}
 

/** 
 * This method must be called before any other method is invoked on this class. Please note that when using ImageLoader as part of  {@link WebImageView} or {@link WebGalleryAdapter}, then there is no need to call this method, since those classes will already do that for you. This method is idempotent. You may call it multiple times without any side effects.
 * @param context the current context
 */
public static synchronized void initialize(Context context){
  if (executor == null) {
    executor=(ThreadPoolExecutor)Executors.newFixedThreadPool(DEFAULT_POOL_SIZE);
  }
  if (imageCache == null) {
    imageCache=new ImageCache(25,expirationInMinutes,DEFAULT_POOL_SIZE);
    imageCache.enableDiskCache(context,ImageCache.DISK_CACHE_SDCARD);
  }
}
 

/** 
 * This method must be called before any other method is invoked on this class. Please note that when using ImageLoader as part of  {@link WebImageView} or {@link }, then there is no need to call this method, since those classes will already do that for you. This method is idempotent. You may call it multiple times without any side effects.
 * @param context the current context
 */
public static synchronized void initialize(Context context){
  if (executor == null) {
    executor=(ThreadPoolExecutor)Executors.newFixedThreadPool(DEFAULT_POOL_SIZE);
  }
  if (imageCache == null) {
    imageCache=new ImageCache(300,expirationInMinutes,DEFAULT_POOL_SIZE);
    imageCache.enableDiskCache(context,ImageCache.DISK_CACHE_SDCARD);
  }
}
 

public OpenflowPipelineFactory(Controller controller,ThreadPoolExecutor pipelineExecutor){
  super();
  this.controller=controller;
  this.pipelineExecutor=pipelineExecutor;
  this.timer=new HashedWheelTimer();
  this.idleHandler=new IdleStateHandler(timer,20,25,0);
  this.readTimeoutHandler=new ReadTimeoutHandler(timer,30);
}
 

public AbstractTcpClient(String name,final Cluster cluster,final String portProperty) throws Exception {
  super(name,cluster);
  this.portProperty=portProperty;
  if (bossExecutor == null)   bossExecutor=(ThreadPoolExecutor)Executors.newCachedThreadPool();
  if (workerExecutor == null)   workerExecutor=(ThreadPoolExecutor)Executors.newCachedThreadPool();
  configureThreadPool(name + "-tcpClientBoss",bossExecutor);
  configureThreadPool(name + "-tcpClientWorker",workerExecutor);
  if (receiveExecutor != null)   configureThreadPool(name + "-tcpClientReceive",receiveExecutor);
  this.channelFactory=new NioClientSocketChannelFactory(bossExecutor,workerExecutor);
  this.bootstrap=new ClientBootstrap(channelFactory);
  origChannelFacotry=new TcpMessagePipelineFactory(LOG,null,receiveExecutor){
    @Override public ChannelPipeline getPipeline() throws Exception {
      final ChannelPipeline pipeline=super.getPipeline();
      pipeline.addBefore("messageCodec","nodeNameWriter",new ChannelNodeNameWriter(cluster));
      pipeline.addBefore("nodeNameWriter","nodeInfoSetter",new SimpleChannelUpstreamHandler(){
        @Override public void channelConnected(        ChannelHandlerContext ctx,        ChannelStateEvent e) throws Exception {
          if (nodeName == null)           throw new RuntimeException("nodeName not set!");
          final NodeInfo ni=cluster.getNodeInfoByName(nodeName);
          ChannelNodeInfo.nodeInfo.set(ctx.getChannel(),ni);
          super.channelConnected(ctx,e);
          pipeline.remove(this);
        }
      }
);
      pipeline.addLast("router",channelHandler);
      return pipeline;
    }
  }
;
  bootstrap.setPipelineFactory(new ChannelPipelineFactory(){
    @Override public ChannelPipeline getPipeline() throws Exception {
      return AbstractTcpClient.this.getPipeline();
    }
  }
);
  bootstrap.setOption("tcpNoDelay",true);
  bootstrap.setOption("keepAlive",true);
  reconnect=true;
}
 

/** 
 * This method must be called before any other method is invoked on this class. Please note that when using ImageLoader as part of  {@link WebImageView} or {@link WebGalleryAdapterOLD}, then there is no need to call this method, since those classes will already do that for you. This method is idempotent. You may call it multiple times without any side effects.
 * @param context the current context
 */
public static synchronized void initialize(Context context){
  if (executor == null) {
    executor=(ThreadPoolExecutor)Executors.newFixedThreadPool(DEFAULT_POOL_SIZE);
  }
  if (imageCache == null) {
    imageCache=new ImageCache(25,DEFAULT_TTL_MINUTES,DEFAULT_POOL_SIZE);
    imageCache.enableDiskCache(context,ImageCache.DISK_CACHE_SDCARD);
  }
}
 

@Override public JobStatus submitJob(BSPJobID jobID,String jobFile) throws IOException {
  this.jobFile=jobFile;
  if (fs == null) {
    this.fs=FileSystem.get(conf);
  }
  conf.addResource(fs.open(new Path(jobFile)));
  conf.setClass(MessageManagerFactory.MESSAGE_MANAGER_CLASS,LocalMessageManager.class,MessageManager.class);
  conf.setClass(SyncServiceFactory.SYNC_CLIENT_CLASS,LocalSyncClient.class,SyncClient.class);
  BSPJob job=new BSPJob(new HamaConfiguration(conf),jobID);
  currentJobStatus=new JobStatus(jobID,System.getProperty("user.name"),0L,JobStatus.RUNNING,globalCounters);
  int numBspTask=job.getNumBspTask();
  String jobSplit=conf.get("bsp.job.split.file");
  BSPJobClient.RawSplit[] splits=null;
  if (jobSplit != null) {
    DataInputStream splitFile=fs.open(new Path(jobSplit));
    try {
      splits=BSPJobClient.readSplitFile(splitFile);
    }
  finally {
      splitFile.close();
    }
  }
  threadPool=(ThreadPoolExecutor)Executors.newFixedThreadPool(numBspTask);
  peerNames=new String[numBspTask];
  for (int i=0; i < numBspTask; i++) {
    peerNames[i]="local:" + i;
    futureList.add(threadPool.submit(new BSPRunner(new Configuration(conf),job,i,splits)));
    globalCounters.incrCounter(JobInProgress.JobCounter.LAUNCHED_TASKS,1L);
  }
  new Thread(new ThreadObserver(currentJobStatus)).start();
  return currentJobStatus;
}
 

public synchronized void start() throws IOException {
  if (isStarted()) {
    return;
  }
  if (getHandler() == null) {
    throw new IOException("The handler is null");
  }
  if (getCodecFactory() == null) {
    setCodecFactory(new ByteBufferCodecFactory());
  }
  setStarted(true);
  setReadEventDispatcher(DispatcherFactory.newDispatcher(getReadThreadCount(),new ThreadPoolExecutor.CallerRunsPolicy(),"xmemcached-read-thread"));
  setWriteEventDispatcher(DispatcherFactory.newDispatcher(getWriteThreadCount(),new ThreadPoolExecutor.CallerRunsPolicy(),"xmemcached-write-thread"));
  setDispatchMessageDispatcher(DispatcherFactory.newDispatcher(getDispatchMessageThreadCount(),new ThreadPoolExecutor.CallerRunsPolicy(),"xmemcached-dispatch-thread"));
  startStatistics();
  start0();
  notifyStarted();
  Runtime.getRuntime().addShutdownHook(new Thread(){
    @Override public void run(){
      try {
        AbstractController.this.stop();
      }
 catch (      IOException e) {
        log.error("Stop controller fail",e);
      }
    }
  }
);
  log.warn("The Controller started at " + localSocketAddress + " ...");
}
 

/** 
 * Creates a new ImageLoader that is backed by an  {@link ImageCache}. The cache will by default cache to the device's external storage, and expire images after 1 day. You can set useCache to false and then supply your own image cache instance via  {@link #setImageCache(ImageCache)}, or fine-tune the default one through  {@link #getImageCache()}.
 * @param context the current context
 * @param createCache whether to create a default  {@link ImageCache} used for caching
 */
public RemoteImageLoader(Context context,boolean createCache){
  executor=(ThreadPoolExecutor)Executors.newFixedThreadPool(DEFAULT_POOL_SIZE);
  if (createCache) {
    imageCache=new ImageCache(25,expirationInMinutes,DEFAULT_POOL_SIZE);
    imageCache.enableDiskCache(context.getApplicationContext(),ImageCache.DISK_CACHE_SDCARD);
  }
  errorDrawable=context.getResources().getDrawable(android.R.drawable.ic_dialog_alert);
}
 

public void onStateChange(SessionState newState,SessionState oldState,Session source){
  if (newState.isBound()) {
    try {
      conn.setSoTimeout(getEnquireLinkTimer());
    }
 catch (    IOException e) {
      logger.error("Failed setting so_timeout for session timer",e);
    }
    logger.info("Changing processor degree to {}",getPduProcessorDegree());
    ((ThreadPoolExecutor)pduReaderWorker.executorService).setCorePoolSize(getPduProcessorDegree());
    ((ThreadPoolExecutor)pduReaderWorker.executorService).setMaximumPoolSize(getPduProcessorDegree());
  }
}
 

public void rejectedExecution(Runnable r,ThreadPoolExecutor e){
  try {
    if (e.isShutdown() || !e.getQueue().offer(r,_time,_timeUnit)) {
      throw new RejectedExecutionException();
    }
  }
 catch (  InterruptedException ie) {
    Thread.currentThread().interrupt();
    throw new RejectedExecutionException(ie);
  }
}
 

@Test public void smoke() throws Exception {
  smtpAppender.setLayout(buildPatternLayout(loggerContext));
  smtpAppender.start();
  Logger logger=loggerContext.getLogger("test");
  logger.addAppender(smtpAppender);
  logger.debug("hello");
  logger.error("en error",new Exception("an exception"));
  waitUntilEmailIsSent();
  System.out.println("*** " + ((ThreadPoolExecutor)loggerContext.getExecutorService()).getCompletedTaskCount());
  List<WiserMessage> wiserMsgList=WISER.getMessages();
  assertNotNull(wiserMsgList);
  assertEquals(numberOfOldMessages + 1,wiserMsgList.size());
  WiserMessage wm=wiserMsgList.get(numberOfOldMessages);
  MimeMessage mm=wm.getMimeMessage();
  assertEquals(TEST_SUBJECT,mm.getSubject());
  MimeMultipart mp=(MimeMultipart)mm.getContent();
  String body=getBody(mp.getBodyPart(0));
  System.out.println("[" + body);
  assertTrue(body.startsWith(HEADER.trim()));
  assertTrue(body.endsWith(FOOTER.trim()));
}
 

/** 
 * This method must be called before any other method is invoked on this class. Please note that when using ImageLoader as part of  {@link WebImageView} or {@link WebGalleryAdapter}, then there is no need to call this method, since those classes will already do that for you. This method is idempotent. You may call it multiple times without any side effects.
 * @param context the current context
 */
public static synchronized void initialize(Context context){
  if (executor == null) {
    executor=(ThreadPoolExecutor)Executors.newFixedThreadPool(DEFAULT_POOL_SIZE);
  }
  if (imageCache == null) {
    imageCache=new ImageCache(25,expirationInMinutes,DEFAULT_POOL_SIZE);
    imageCache.enableDiskCache(context,ImageCache.DISK_CACHE_SDCARD);
  }
}
 

/** 
 * stop synchronously, handy for unit test scenarios.
 * @param delaySeconds
 * @return
 */
public JerseyServer stop(int delaySeconds){
  server.stop(delaySeconds);
  Executor serverExecutor=server.getExecutor();
  if (serverExecutor instanceof ThreadPoolExecutor) {
    ((ThreadPoolExecutor)serverExecutor).shutdown();
    if (delaySeconds > 0) {
      try {
        ((ThreadPoolExecutor)serverExecutor).awaitTermination(delaySeconds,TimeUnit.SECONDS);
      }
 catch (      InterruptedException ex) {
      }
    }
  }
  return this;
}
 

/** 
 * Converts the given ThreadPoolExecutor into an ExecutorService that exits when the application is complete.  It does so by using daemon threads and adding a shutdown hook to wait for their completion. <p>This is mainly for fixed thread pools. See  {@link java.util.concurrent.Executors#newFixedThreadPool(int)}.
 * @param executor the executor to modify to make sure it exits when theapplication is finished
 * @param terminationTimeout how long to wait for the executor tofinish before terminating the JVM
 * @param timeUnit unit of time for the time parameter
 * @return an unmodifiable version of the input which will not hang the JVM
 */
public static ExecutorService getExitingExecutorService(ThreadPoolExecutor executor,long terminationTimeout,TimeUnit timeUnit){
  executor.setThreadFactory(daemonThreadFactory(executor.getThreadFactory()));
  ExecutorService service=java.util.concurrent.Executors.unconfigurableExecutorService(executor);
  addDelayedShutdownHook(service,terminationTimeout,timeUnit);
  return service;
}
 

public void rejectedExecution(Runnable r,ThreadPoolExecutor executor){
  if (null == queue) {
    queue=executor.getQueue();
  }
  try {
    queue.put(r);
  }
 catch (  InterruptedException e) {
    Thread.interrupted();
    throw new RejectedExecutionException(e);
  }
}
 

private void initRemotingClient(long connectTimeoutInMills) throws IOException {
  final ClientConfig clientConfig=new ClientConfig();
  clientConfig.setWireFormatType(new PushitWireFormatType());
  clientConfig.setConnectTimeout(connectTimeoutInMills);
  remotingClient=RemotingFactory.newRemotingClient(clientConfig);
  remotingClient.registerProcessor(NotifyCommand.class,new RequestProcessor<NotifyCommand>(){
    public ThreadPoolExecutor getExecutor(){
      return null;
    }
    public void handleRequest(    NotifyCommand request,    Connection conn){
      if (notifyListener != null) {
        notifyListener.onNotify(request.getDataId(),request.getGroup(),request.getMessage());
      }
    }
  }
);
  remotingClient.addConnectionLifeCycleListener(new ReconnectionListener());
  try {
    remotingClient.start();
  }
 catch (  NotifyRemotingException e) {
    throw new IOException(e);
  }
}
 

public OpenflowPipelineFactory(Controller controller,ThreadPoolExecutor pipelineExecutor){
  super();
  this.controller=controller;
  this.pipelineExecutor=pipelineExecutor;
  this.timer=new HashedWheelTimer();
  this.idleHandler=new IdleStateHandler(timer,20,25,0);
  this.readTimeoutHandler=new ReadTimeoutHandler(timer,30);
}
 

public synchronized void start() throws IOException {
  if (isStarted()) {
    return;
  }
  if (getHandler() == null) {
    throw new IOException("The handler is null");
  }
  if (getCodecFactory() == null) {
    setCodecFactory(new ByteBufferCodecFactory());
  }
  setStarted(true);
  setReadEventDispatcher(DispatcherFactory.newDispatcher(getReadThreadCount(),new ThreadPoolExecutor.CallerRunsPolicy(),"xmemcached-read-thread"));
  setWriteEventDispatcher(DispatcherFactory.newDispatcher(getWriteThreadCount(),new ThreadPoolExecutor.CallerRunsPolicy(),"xmemcached-write-thread"));
  setDispatchMessageDispatcher(DispatcherFactory.newDispatcher(getDispatchMessageThreadCount(),new ThreadPoolExecutor.CallerRunsPolicy(),"xmemcached-dispatch-thread"));
  startStatistics();
  start0();
  notifyStarted();
  shutdownHookThread=new Thread(){
    @Override public void run(){
      try {
        isHutdownHookCalled=true;
        AbstractController.this.stop();
      }
 catch (      IOException e) {
        log.error("Stop controller fail",e);
      }
    }
  }
;
  Runtime.getRuntime().addShutdownHook(shutdownHookThread);
  log.warn("The Controller started at " + localSocketAddress + " ...");
}
 

/** 
 * Construct a new instance.  Intended to be called only from implementations.  To construct an XNIO worker, use the  {@link Xnio#createWorker(OptionMap)} method.
 * @param xnio the XNIO provider which produced this worker instance
 * @param threadGroup the thread group for worker threads
 * @param optionMap the option map to use to configure this worker
 * @param terminationTask
 */
protected XnioWorker(final Xnio xnio,final ThreadGroup threadGroup,final OptionMap optionMap,final Runnable terminationTask){
  this.xnio=xnio;
  this.terminationTask=terminationTask;
  String workerName=optionMap.get(Options.WORKER_NAME);
  if (workerName == null) {
    workerName="XNIO-" + seq.getAndIncrement();
  }
  name=workerName;
  BlockingQueue<Runnable> taskQueue=new LinkedTransferQueue<Runnable>();
  this.coreSize=optionMap.get(Options.WORKER_TASK_CORE_THREADS,4);
  final boolean markThreadAsDaemon=optionMap.get(Options.THREAD_DAEMON,false);
  taskPool=new TaskPool(optionMap.get(Options.WORKER_TASK_MAX_THREADS,16),optionMap.get(Options.WORKER_TASK_MAX_THREADS,16),optionMap.get(Options.WORKER_TASK_KEEPALIVE,60),TimeUnit.MILLISECONDS,taskQueue,new ThreadFactory(){
    public Thread newThread(    final Runnable r){
      final Thread taskThread=new Thread(threadGroup,r,name + " task-" + getNextSeq(),optionMap.get(Options.STACK_SIZE,0L));
      if (markThreadAsDaemon) {
        taskThread.setDaemon(true);
      }
      return taskThread;
    }
  }
,new ThreadPoolExecutor.AbortPolicy());
}