Java Code Examples for java.util.concurrent.ThreadFactory

/** 
 * Creates the executor service used to run tasks.
 * @param numTasks the total number of map tasks to be run
 * @return an ExecutorService instance that handles map tasks
 */
protected ExecutorService createTaskExecutor(int numTasks){
  int maxThreads=conf.getInt(LOCAL_MAX_TASKS,1);
  if (maxThreads < 1) {
    throw new IllegalArgumentException("Configured " + LOCAL_MAX_TASKS + " must be >= 1");
  }
  maxThreads=Math.min(maxThreads,numTasks);
  maxThreads=Math.max(maxThreads,1);
  LOG.debug("Starting thread pool executor.");
  LOG.debug("Max local threads: " + maxThreads);
  LOG.debug("Tasks to process: " + numTasks);
  ThreadFactory tf=new ThreadFactoryBuilder().setNameFormat("LocalJobRunner Task Executor #%d").build();
  ExecutorService executor=Executors.newFixedThreadPool(maxThreads,tf);
  return executor;
}
 

@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 void start(){
  if (this.ses != null && !this.ses.isShutdown() && !this.ses.isTerminated()) {
    return;
  }
  final ThreadFactory thread=new ThreadFactory(){
    private final AtomicInteger threadNumber=new AtomicInteger(1);
    @Override public Thread newThread(    final Runnable r){
      final Thread t=new Thread(r);
      t.setName("Timermanager-Worker-" + this.threadNumber.getAndIncrement());
      return t;
    }
  }
;
  final ScheduledThreadPoolExecutor stpe=new ScheduledThreadPoolExecutor(3,thread);
  stpe.setKeepAliveTime(10,TimeUnit.MINUTES);
  stpe.allowCoreThreadTimeOut(true);
  stpe.setCorePoolSize(3);
  stpe.setContinueExistingPeriodicTasksAfterShutdownPolicy(false);
  this.ses=stpe;
}
 

HintCaskOutput(final File path) throws IOException {
  if (path.exists()) {
    throw new IOException(path + " already exists");
  }
  FileOutputStream fos=new FileOutputStream(path);
  m_fc=fos.getChannel();
  m_fc.position(4);
  final ThreadFactory tf=new ThreadFactory(){
    @Override public Thread newThread(    Runnable r){
      Thread t=new Thread(null,r,"HintCask[" + path + "] write thread",1024 * 256);
      t.setDaemon(true);
      return t;
    }
  }
;
  m_thread=MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor(tf));
}
 

protected Factory<Executor> createExecutorFactory(final int numThreads){
  final ThreadFactory threadFactory=getQueryThreadFactory();
  return new Factory<Executor>(){
    public Executor create(){
      return Executors.newFixedThreadPool(numThreads,threadFactory);
    }
  }
;
}
 

private ThreadFactory getThreadFactory(CuratorFrameworkFactory.Builder builder){
  ThreadFactory threadFactory=builder.getThreadFactory();
  if (threadFactory == null) {
    threadFactory=ThreadUtils.newThreadFactory("CuratorFramework");
  }
  return threadFactory;
}
 

/** 
 * @param hive
 * @param prefix
 */
public ProxyGroup(final Hive hive,final ProxyPrefixName prefix){
  this.hive=hive;
  this.prefix=prefix;
  if (Boolean.getBoolean("org.openscada.da.server.proxy.asyncListener")) {
    final ThreadFactory tf=new ThreadFactoryImplementation(prefix.getName());
    this.itemListenerExecutor=Executors.newSingleThreadExecutor(tf);
  }
}
 

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;
}
 

private void startManager(){
  ThreadFactory threadFactory=new ThreadFactory(){
    private int i=0;
    public Thread newThread(    Runnable r){
      return new Thread(r,"Delivery-Manager-" + this.i++);
    }
  }
;
  this.managerExecuter=Executors.newCachedThreadPool(threadFactory);
  for (int i=0; i < this.managerThreads; i++) {
    this.managerExecuter.submit(new Worker());
  }
}
 

public PodcastService(){
  ThreadFactory threadFactory=new ThreadFactory(){
    public Thread newThread(    Runnable r){
      Thread t=Executors.defaultThreadFactory().newThread(r);
      t.setDaemon(true);
      return t;
    }
  }
;
  refreshExecutor=Executors.newFixedThreadPool(5,threadFactory);
  downloadExecutor=Executors.newFixedThreadPool(3,threadFactory);
  scheduledExecutor=Executors.newSingleThreadScheduledExecutor(threadFactory);
}
 

public PodcastService(){
  ThreadFactory threadFactory=new ThreadFactory(){
    public Thread newThread(    Runnable r){
      Thread t=Executors.defaultThreadFactory().newThread(r);
      t.setDaemon(true);
      return t;
    }
  }
;
  refreshExecutor=Executors.newFixedThreadPool(5,threadFactory);
  downloadExecutor=Executors.newFixedThreadPool(3,threadFactory);
  scheduledExecutor=Executors.newSingleThreadScheduledExecutor(threadFactory);
}
 

public PodcastService(){
  ThreadFactory threadFactory=new ThreadFactory(){
    public Thread newThread(    Runnable r){
      Thread t=Executors.defaultThreadFactory().newThread(r);
      t.setDaemon(true);
      return t;
    }
  }
;
  refreshExecutor=Executors.newFixedThreadPool(5,threadFactory);
  downloadExecutor=Executors.newFixedThreadPool(3,threadFactory);
  scheduledExecutor=Executors.newSingleThreadScheduledExecutor(threadFactory);
}
 

public PodcastService(){
  ThreadFactory threadFactory=new ThreadFactory(){
    public Thread newThread(    Runnable r){
      Thread t=Executors.defaultThreadFactory().newThread(r);
      t.setDaemon(true);
      return t;
    }
  }
;
  refreshExecutor=Executors.newFixedThreadPool(5,threadFactory);
  downloadExecutor=Executors.newFixedThreadPool(3,threadFactory);
  scheduledExecutor=Executors.newSingleThreadScheduledExecutor(threadFactory);
}
 

public static ExecutorService newExecutor(){
  ThreadFactory jobThreadFactory=new ThreadFactory(){
    ThreadFactory defaultThreadFactory=Executors.defaultThreadFactory();
    @Override public Thread newThread(    Runnable r){
      Thread newThread=defaultThreadFactory.newThread(r);
      newThread.setUncaughtExceptionHandler(loggingUEH);
      newThread.setContextClassLoader(this.getClass().getClassLoader());
      return newThread;
    }
  }
;
  return Executors.newCachedThreadPool(jobThreadFactory);
}
 

public Comptroller(){
  ThreadFactory tf=new ThreadFactory(){
    @Override public Thread newThread(    Runnable r){
      Thread t=new Thread(r,"Sensor thread");
      return t;
    }
  }
;
  this.exec=new ScheduledThreadPoolExecutor(1,tf);
  this.sensors=new ArrayList<>();
}
 

protected void initExecutor(Map<String,Object> config){
  if (executor == null) {
    int threads=ConfigUtils.getInteger(config,MAX_POOL_SIZE,CFG_PREFIX + ".threads");
    if (threads <= 1) {
      executor=new Executor(){
        public void execute(        Runnable command){
          command.run();
        }
      }
;
    }
 else {
      ThreadFactory threadFactory=new RepositoryConnectorThreadFactory(getClass().getSimpleName());
      executor=new ThreadPoolExecutor(threads,threads,3,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(),threadFactory);
    }
  }
}
 

private static ScheduledExecutorService createExecutor(){
  return Executors.newSingleThreadScheduledExecutor(new ThreadFactory(){
    public Thread newThread(    Runnable r){
      Thread t=new Thread(r,"akubra-rmi-unexporter");
      t.setPriority(Thread.MIN_PRIORITY);
      t.setDaemon(true);
      return t;
    }
  }
);
}
 

/** 
 * This method is implemented with  {@link java.util.concurrent.ScheduledExecutorService#scheduleWithFixedDelay(Runnable,long,long,TimeUnit)}
 */
@Override protected synchronized void schedule(Runnable runnable){
  executor=Executors.newScheduledThreadPool(1,new ThreadFactory(){
    @Override public Thread newThread(    Runnable r){
      Thread t=new Thread(r,"pollingConfigurationSource");
      t.setDaemon(true);
      return t;
    }
  }
);
  executor.scheduleWithFixedDelay(runnable,initialDelayMillis,delayMillis,TimeUnit.MILLISECONDS);
}
 

private static ThreadFactory createThreadFactory(final String format,final AtomicLong threadPoolCounter){
  return new ThreadFactory(){
    public Thread newThread(    Runnable runnable){
      Thread thread=new Thread(runnable);
      thread.setName(String.format(format,threadPoolCounter.getAndIncrement()));
      return thread;
    }
  }
;
}
 

/** 
 * Utility method for creating a cached pool of "daemon" threads.  A daemon thread does not limit the JVM from exiting if they aren't shutdown.
 * @return A new cached pool of daemon threads
 */
static public ExecutorService newCachedDaemonThreadPool(){
  return Executors.newCachedThreadPool(new ThreadFactory(){
    @Override public Thread newThread(    Runnable r){
      Thread t=new Thread(r);
      t.setDaemon(true);
      return t;
    }
  }
);
}
 

/** 
 */
public String getPacketStats(){
  TextBuilder tb=new TextBuilder();
  ThreadFactory tf=_generalPacketsThreadPool.getThreadFactory();
  if (tf instanceof PriorityThreadFactory) {
    tb.append("General Packet Thread Pool:\r\n");
    tb.append("Tasks in the queue: " + _generalPacketsThreadPool.getQueue().size() + "\r\n");
    tb.append("Showing threads stack trace:\r\n");
    PriorityThreadFactory ptf=(PriorityThreadFactory)tf;
    int count=ptf.getGroup().activeCount();
    Thread[] threads=new Thread[count + 2];
    ptf.getGroup().enumerate(threads);
    tb.append("There should be " + count + " Threads\r\n");
    for (    Thread t : threads) {
      if (t == null) {
        continue;
      }
      tb.append(t.getName() + "\r\n");
      for (      StackTraceElement ste : t.getStackTrace()) {
        tb.append(ste.toString());
        tb.append("\r\n");
      }
    }
  }
  tb.append("Packet Tp stack traces printed.\r\n");
  return tb.toString();
}
 

@Override ExecutorService createExecutorService(){
  return Executors.newSingleThreadExecutor(new ThreadFactory(){
    @Override public Thread newThread(    Runnable run){
      Thread thread=new Thread(run);
      thread.setName("OSGi PackageAdmin refresh Thread");
      thread.setDaemon(true);
      return thread;
    }
  }
);
}
 

ContainerExecutor(final int corePoolSize,final int maximumPoolSize,final long keepAliveTime,final TimeUnit unit){
  super(corePoolSize,maximumPoolSize,keepAliveTime,unit,new LinkedBlockingQueue<Runnable>(),new ThreadFactory(){
    private final int id=executorSeq.getAndIncrement();
    private final AtomicInteger threadSeq=new AtomicInteger(1);
    public Thread newThread(    final Runnable r){
      Thread thread=new ServiceThread(r,ServiceContainerImpl.this);
      thread.setName(String.format("MSC service thread %d-%d",Integer.valueOf(id),Integer.valueOf(threadSeq.getAndIncrement())));
      thread.setUncaughtExceptionHandler(HANDLER);
      return thread;
    }
  }
,POLICY);
}
 

/** 
 * Startup all workers.
 */
public void startup(){
  final String[] workerScript=getWorkerScript();
  if (workerScript == null) {
    final String message="no worker script to execute - configure one using '" + SCRIPT_KEY + "' "+ "or '"+ SCRIPT_PATH_KEY+ "' parameter (or see previous errors if already configured) ";
    log("[" + getClass().getName() + "] "+ message+ " !");
    return;
  }
  final int workersCount=getThreadCount();
  final ThreadFactory threadFactory=newThreadFactory();
  for (int i=0; i < workersCount; i++) {
    final Ruby runtime=getRuntime();
    if (isExported()) {
      runtime.getGlobalVariables().set(GLOBAL_VAR_NAME,JavaEmbedUtils.javaToRuby(runtime,this));
    }
    try {
      final RubyWorker worker=newRubyWorker(runtime,workerScript[0],workerScript[1]);
      final Thread workerThread=threadFactory.newThread(worker);
      workers.put(worker,workerThread);
      workerThread.start();
    }
 catch (    Exception e) {
      log("[" + getClass().getName() + "] worker startup failed",e);
      break;
    }
  }
  log("[" + getClass().getName() + "] started "+ workers.size()+ " worker(s)");
}
 

public ChannelFactory(String host,int port){
  this.host=host;
  this.port=port;
  ThreadFactory tf=new ThreadFactory(){
    @Override public Thread newThread(    Runnable r){
      Thread t=new Thread(r);
      t.setDaemon(true);
      return t;
    }
  }
;
  final ExecutorService threadPool=Executors.newCachedThreadPool(tf);
  bootstrap=new ClientBootstrap(new OioClientSocketChannelFactory(threadPool));
  bootstrap.setPipelineFactory(new PipelineFactory());
}
 

private static ExecutorService getExecutor(){
  if (sBitmapFileCleanerExecutor == null) {
    sBitmapFileCleanerExecutor=Executors.newSingleThreadExecutor(new ThreadFactory(){
      @Override public Thread newThread(      Runnable r){
        Thread thread=new Thread(r);
        thread.setName(TAG + " | " + thread.getName());
        thread.setPriority(Thread.MIN_PRIORITY);
        return thread;
      }
    }
);
  }
  return sBitmapFileCleanerExecutor;
}
 

/** 
 * Constructor
 */
public ThreadPerTaskExecutor(){
  this(new ThreadFactory(){
    @Override public Thread newThread(    Runnable r){
      return new Thread(r);
    }
  }
);
}
 

public EJB3ExecutorService(){
  executorService=Executors.newCachedThreadPool(new ThreadFactory(){
    @Override public Thread newThread(    Runnable r){
      return new EJB3ThreadWorker(r);
    }
  }
);
}
 

private static synchronized ExecutorService getExecutorService(){
  if (executorInstance == null) {
    executorInstance=Executors.newCachedThreadPool(new ThreadFactory(){
      @Override public Thread newThread(      Runnable r){
        Thread t=new Thread(r);
        t.setDaemon(true);
        return t;
      }
    }
);
  }
  return executorInstance;
}
 

public AsyncAutomationClient(String url){
  this(url,Executors.newCachedThreadPool(new ThreadFactory(){
    public Thread newThread(    Runnable r){
      return new Thread("AutomationAsyncExecutor");
    }
  }
));
}
 

public void start(){
  this.flushThread=new FlushThread();
  this.scheduledExecutor=Executors.newSingleThreadScheduledExecutor(new ThreadFactory(){
    @Override public Thread newThread(    Runnable r){
      Thread t=new Thread(Thread.currentThread().getThreadGroup(),r);
      t.setDaemon(true);
      t.setName("Flush Thread");
      return t;
    }
  }
);
  this.flushFuture=scheduledExecutor.schedule(flushThread,TSOState.FLUSH_TIMEOUT,TimeUnit.MILLISECONDS);
  this.executor=Executors.newSingleThreadExecutor();
}
 

public void contextInitialized(ServletContextEvent arg0){
  ServletContext context=arg0.getServletContext();
  int nr_executors=2;
  ThreadFactory daemonFactory=new DaemonThreadFactory();
  try {
    nr_executors=Integer.parseInt(context.getInitParameter("nr-executors"));
  }
 catch (  NumberFormatException ignore) {
  }
  if (nr_executors <= 1) {
    executor=Executors.newSingleThreadExecutor(daemonFactory);
  }
 else {
    executor=Executors.newFixedThreadPool(nr_executors,daemonFactory);
  }
  context.setAttribute("MY_EXECUTOR",executor);
}
 

/** 
 * Creates the  {@link ExecutorService} for writing to pipes.<p> If the throttling is supported, use a separate thread to free up the main channel reader thread (thus prevent blockage.) Otherwise let the channel reader thread do it, which is the historical behaviour.
 */
private ExecutorService createPipeWriterExecutor(){
  if (remoteCapability.supportsPipeThrottling())   return Executors.newSingleThreadExecutor(new ThreadFactory(){
    public Thread newThread(    Runnable r){
      return new Thread(r,"Pipe writer thread: " + name);
    }
  }
);
  return new SynchronousExecutorService();
}
 

/** 
 * Creates a single thread  {@link ScheduledExecutorService} to validate sessions at fixed intervals and enables this scheduler. The executor is created as a daemon thread to allow JVM to shut down
 */
public void enableSessionValidation(){
  if (this.interval > 0l) {
    this.service=Executors.newSingleThreadScheduledExecutor(new ThreadFactory(){
      public Thread newThread(      Runnable r){
        Thread thread=new Thread(r);
        thread.setDaemon(true);
        return thread;
      }
    }
);
    this.service.scheduleAtFixedRate(this,interval,interval,TimeUnit.MILLISECONDS);
    this.enabled=true;
  }
}
 

private static synchronized ExecutorService getExecutorService(){
  if (executorInstance == null) {
    executorInstance=Executors.newCachedThreadPool(new ThreadFactory(){
      @Override public Thread newThread(      Runnable r){
        Thread t=new Thread(r);
        t.setDaemon(true);
        return t;
      }
    }
);
  }
  return executorInstance;
}
 

protected void initExecutor(Map<String,Object> config){
  if (executor == null) {
    int threads=ConfigUtils.getInteger(config,MAX_POOL_SIZE,CFG_PREFIX + ".threads");
    if (threads <= 1) {
      executor=new Executor(){
        public void execute(        Runnable command){
          command.run();
        }
      }
;
    }
 else {
      ThreadFactory threadFactory=new RepositoryConnectorThreadFactory(getClass().getSimpleName());
      executor=new ThreadPoolExecutor(threads,threads,3,TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(),threadFactory);
    }
  }
}
 

@Override protected void runServer(){
  ExecutorService threadPool=null;
  try {
    socket=new TcpServerServerSocket(this,port);
    port=socket.getLocalPort();
    socket.setSoTimeout(0);
    callOnServerStarted();
    ThreadFactory factory=new DaemonThreadFactory(new RenamingThreadFactory(Thread.currentThread().getName() + "-HandlerThread-"),daemon);
    threadPool=Executors.newCachedThreadPool(factory);
    while (!Thread.currentThread().isInterrupted()) {
      try {
        threadPool.execute(new ConnectionHandler(this,socket.accept()));
      }
 catch (      IOException e) {
        if (!Thread.currentThread().isInterrupted())         callOnIOException(null,e,"Error while receiving");
      }
    }
  }
 catch (  BindException e) {
    thread=null;
    callOnIOException(null,e,"Can't bind port");
  }
catch (  IOException e) {
    thread=null;
    callOnIOException(null,e,"Couldn't open ServerSocket on port " + port + "!");
  }
 finally {
    if (threadPool != null)     threadPool.shutdown();
    callOnServerStopped();
  }
}
 

/** 
 * Creates a new scheduled thread pool which can schedule threads.
 * @param poolSize the core pool size
 * @param pattern  pattern of the thread name
 * @param name     ${name} in the pattern name
 * @param daemon   whether the threads is daemon or not
 * @return the created pool
 */
public static ScheduledExecutorService newScheduledThreadPool(final int poolSize,final String pattern,final String name,final boolean daemon){
  return Executors.newScheduledThreadPool(poolSize,new ThreadFactory(){
    public Thread newThread(    Runnable r){
      Thread answer=new Thread(r,getThreadName(pattern,name));
      answer.setDaemon(daemon);
      return answer;
    }
  }
);
}
 

public ContextPropagatingTaskExecutor(final String threadNamePrefix,int poolSize,BundleContext bundleContext){
  this.bundleContext=bundleContext;
  this.executor=Executors.newFixedThreadPool(poolSize,new ThreadFactory(){
    public Thread newThread(    Runnable r){
      Thread t=new Thread(r);
      t.setName(threadNamePrefix + ContextPropagatingTaskExecutor.this.threadCount.getAndIncrement());
      return t;
    }
  }
);
}
 

private static synchronized ExecutorService getExecutorService(){
  if (executorInstance == null) {
    executorInstance=Executors.newCachedThreadPool(new ThreadFactory(){
      public Thread newThread(      Runnable r){
        Thread t=new Thread(r);
        t.setDaemon(true);
        return t;
      }
    }
);
  }
  return executorInstance;
}
 

/** 
 * @param name
 * @param password
 * @param connectionTimeout
 * @throws IOException
 */
public TwitterStreamingConnection(String name,String password,int connectionTimeout) throws IOException {
  httpClient=new HttpClient();
  httpClient.getHttpConnectionManager().getParams().setConnectionTimeout(connectionTimeout);
  httpClient.getHttpConnectionManager().getParams().setSoTimeout(10 * 1000);
  httpClient.getParams().setAuthenticationPreemptive(true);
  httpClient.getState().setCredentials(AuthScope.ANY,new UsernamePasswordCredentials(name,password));
  doOpen();
  Executors.newSingleThreadExecutor(new ThreadFactory(){
    @Override public Thread newThread(    Runnable runnable){
      return new Thread(runnable,"TwitterStreamingConnection");
    }
  }
).execute(new Runnable(){
    @Override public void run(){
      BlockingQueue<String> queue=TwitterStreamingConnection.this.queue;
      String line;
      while ((line=readLine()) != null) {
        queue.add(line);
      }
    }
  }
);
}
 

private static synchronized void detectCompletionConcurrently(ZooKeeperPlus zk,String path,Collection<String> relPaths,CompletionAwaiter awaiter,CompletionDetector completionDetector){
  final ExecutorService completionDetectionExecutor=Executors.newFixedThreadPool(NUM_CONCURRENT_COMPLETION_DETECTORS,new ThreadFactory(){
    private int threadId=0;
    @Override public Thread newThread(    Runnable runnable){
      return new Thread(runnable,"Completion Detector #" + threadId++);
    }
  }
);
  for (  String relPath : relPaths) {
    completionDetectionExecutor.execute(new DetectCompletionRunnable(zk,path,relPath,awaiter,completionDetector));
  }
  completionDetectionExecutor.shutdown();
  boolean terminated=false;
  while (!terminated) {
    try {
      terminated=completionDetectionExecutor.awaitTermination(COMPLETION_DETECTION_EXECUTOR_TERMINATION_CHECK_PERIOD,TimeUnit.MILLISECONDS);
    }
 catch (    InterruptedException e) {
      completionDetectionExecutor.shutdownNow();
    }
  }
}
 

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);
}
 

private void runInNewThread(final IBehaviorResults results,final Specification<?> spec){
  ExecutorService executor=Executors.newSingleThreadExecutor(new ThreadFactory(){
    public Thread newThread(    Runnable r){
      return new Thread(r);
    }
  }
);
  executor.submit(new Callable<Void>(){
    public Void call() throws Exception {
      runSpec(results,spec);
      return null;
    }
  }
);
  executor.shutdown();
  try {
    executor.awaitTermination(Long.MAX_VALUE,TimeUnit.SECONDS);
  }
 catch (  InterruptedException e) {
    throw new RuntimeException(e);
  }
}
 

public MailServiceImpl(InitParams params,final ExoContainerContext ctx) throws Exception {
  props_=new Properties(PrivilegedSystemHelper.getProperties());
  props_.putAll(params.getPropertiesParam("config").getProperties());
  if ("true".equals(props_.getProperty("mail.smtp.auth"))) {
    String username=props_.getProperty("mail.smtp.auth.username");
    String password=props_.getProperty("mail.smtp.auth.password");
    final ExoAuthenticator auth=new ExoAuthenticator(username,password);
    mailSession_=SecurityHelper.doPrivilegedAction(new PrivilegedAction<Session>(){
      public Session run(){
        return Session.getInstance(props_,auth);
      }
    }
);
  }
 else {
    mailSession_=SecurityHelper.doPrivilegedAction(new PrivilegedAction<Session>(){
      public Session run(){
        return Session.getInstance(props_,null);
      }
    }
);
  }
  int threadNumber=props_.getProperty(MAX_THREAD_NUMBER) != null ? Integer.valueOf(props_.getProperty(MAX_THREAD_NUMBER)) : Runtime.getRuntime().availableProcessors();
  executorService=Executors.newFixedThreadPool(threadNumber,new ThreadFactory(){
    public Thread newThread(    Runnable arg0){
      return new Thread(arg0,ctx.getName() + "-MailServiceThread-" + mailServiceThreadCounter++);
    }
  }
);
}
 

@Override public ThreadPool createThreadPool(final String name,final int threads,final long threadTimeout,final TimeUnit threadTimeoutUnit,final ThreadPoolPriority priority){
  log.debug("Creating new {} priority ThreadPool {}; threads: {}, timeout:{}",new Object[]{priority,name,threads,threadTimeout});
  final ScheduledThreadPoolExecutor executor=new ScheduledThreadPoolExecutor(threads);
  if (threadTimeout >= 1) {
    executor.setKeepAliveTime(threadTimeout,threadTimeoutUnit);
    executor.allowCoreThreadTimeOut(true);
  }
  executor.setThreadFactory(new ThreadFactory(){
    private final AtomicInteger threadNumber=new AtomicInteger(1);
    @Override public Thread newThread(    Runnable r){
      final Thread thread=new Thread(r,name + "-" + threadNumber.getAndIncrement());
      thread.setPriority(priority.threadPriority);
      return thread;
    }
  }
);
  final ThreadPoolImpl pool=new ThreadPoolImpl(name,executor);
  threadPools.put(name,pool);
  return pool;
}
 

static void enableJitting(boolean enable){
  if (enable) {
    jittingEnabled=true;
    if (executor == null) {
      executor=Executors.newCachedThreadPool(new ThreadFactory(){
        public Thread newThread(        Runnable r){
          Thread t=new Thread(r);
          t.setDaemon(true);
          return t;
        }
      }
);
    }
  }
 else {
    jittingEnabled=false;
    if (executor != null) {
      executor.shutdown();
      executor=null;
    }
  }
}
 

/** 
 * {@inheritDoc}
 */
protected void afterStartApplication(final Object startupObject){
  String strDelay=getWebBeansContext().getOpenWebBeansConfiguration().getProperty(OpenWebBeansConfiguration.CONVERSATION_PERIODIC_DELAY,"150000");
  long delay=Long.parseLong(strDelay);
  service=Executors.newScheduledThreadPool(1,new ThreadFactory(){
    public Thread newThread(    Runnable runable){
      Thread t=new Thread(runable,"OwbConversationCleaner-" + ServletCompatibilityUtil.getServletInfo((ServletContext)(startupObject)));
      t.setDaemon(true);
      return t;
    }
  }
);
  service.scheduleWithFixedDelay(new ConversationCleaner(),delay,delay,TimeUnit.MILLISECONDS);
  ELAdaptor elAdaptor=getWebBeansContext().getService(ELAdaptor.class);
  ELResolver resolver=elAdaptor.getOwbELResolver();
  if (getWebBeansContext().getOpenWebBeansConfiguration().isJspApplication()) {
    logger.log(Level.FINE,"Application is configured as JSP. Adding EL Resolver.");
    JspFactory factory=JspFactory.getDefaultFactory();
    if (factory != null) {
      JspApplicationContext applicationCtx=factory.getJspApplicationContext((ServletContext)(startupObject));
      applicationCtx.addELResolver(resolver);
    }
 else {
      logger.log(Level.FINE,"Default JSPFactroy instance has not found");
    }
  }
  ServletContext servletContext=(ServletContext)(startupObject);
  servletContext.setAttribute(BeanManager.class.getName(),getBeanManager());
}
 

/** 
 * Create a new bundle watcher.
 * @param context   a bundle context. Cannot be null.
 * @param scanner   a bundle scanner. Cannot be null.
 * @param observers list of observers
 */
public BundleWatcher(final BundleContext context,final BundleScanner<T> scanner,final BundleObserver<T>... observers){
  LOG.debug("Creating bundle watcher with scanner [" + scanner + "]...");
  NullArgumentException.validateNotNull(context,"Context");
  NullArgumentException.validateNotNull(scanner,"Bundle scanner");
  m_context=context;
  m_scanner=scanner;
  m_observers=new ArrayList<BundleObserver<T>>();
  if (observers != null) {
    m_observers.addAll(Arrays.asList(observers));
  }
  executorService=Executors.newScheduledThreadPool(3,new ThreadFactory(){
    private final AtomicInteger count=new AtomicInteger();
    public Thread newThread(    Runnable r){
      final Thread t=Executors.defaultThreadFactory().newThread(r);
      t.setName("BundleWatcher" + ": " + count.incrementAndGet());
      t.setDaemon(true);
      return t;
    }
  }
);
}
 

/** 
 * Starts an web.xml watcher on installed bundles.
 * @see BundleActivator#start(BundleContext)
 */
@SuppressWarnings("unchecked") public void start(final BundleContext bundleContext) throws Exception {
  LOG.debug("Pax Web WAR Extender - Starting");
  this.bundleContext=bundleContext;
  executors=Executors.newScheduledThreadPool(3,new ThreadFactory(){
    private final AtomicInteger count=new AtomicInteger();
    public Thread newThread(    Runnable r){
      final Thread t=Executors.defaultThreadFactory().newThread(r);
      t.setName("WebListenerExecutor" + ": " + count.incrementAndGet());
      t.setDaemon(true);
      return t;
    }
  }
);
  webEventDispatcher=new WebEventDispatcher(bundleContext,executors);
  Filter filterEvent=bundleContext.createFilter("(objectClass=org.osgi.service.event.EventAdmin)");
  eventServiceTracker=new ServiceTracker(bundleContext,filterEvent,new EventServiceCustomizer());
  eventServiceTracker.open();
  Filter filterLog=bundleContext.createFilter("(objectClass=org.osgi.service.log.LogService)");
  logServiceTracker=new ServiceTracker(bundleContext,filterLog,new LogServiceCustomizer());
  logServiceTracker.open();
  DefaultWebAppDependencyManager dependencyManager=new DefaultWebAppDependencyManager(bundleContext);
  webXmlObserver=new WebXmlObserver(new DOMWebXmlParser(),new WebAppPublisher(),webEventDispatcher,dependencyManager,bundleContext);
  m_webXmlWatcher=new BundleWatcher<URL>(bundleContext,new BundleURLScanner("Webapp-Root",null,null,"WEB-INF/","*web*.xml",true),webXmlObserver);
  m_webXmlWatcher.start();
  httpServiceTracker=new ReplaceableService<HttpService>(bundleContext,HttpService.class,dependencyManager);
  httpServiceTracker.start();
  bundleContext.registerService(new String[]{WarManager.class.getName()},webXmlObserver,new Hashtable());
  LOG.debug("Pax Web WAR Extender - Started");
}
 

ClientExecutorManager(final Map<String,?> environment){
  if (environment != null && environment.containsKey(Executor.class.getName())) {
    executor=(Executor)environment.get(Executor.class.getName());
  }
 else {
    executor=Executors.newCachedThreadPool(new ThreadFactory(){
      final ThreadGroup group=new ThreadGroup(REMOTING_JMX);
      final AtomicInteger threadNumber=new AtomicInteger(1);
      public Thread newThread(      Runnable r){
        return new Thread(group,r,REMOTING_JMX + " " + CLIENT_THREAD+ threadNumber.getAndIncrement());
      }
    }
);
    manageExecutor=true;
  }
}
 

protected void setUp() throws Exception {
  contexts=new Contexts();
  scheduler=mock(Scheduler.class);
  contexts.scheduler=(Scheduler)scheduler.proxy();
  cronScheduler=new CronScheduler();
  cronScheduler.setContexts(contexts);
  execService=Executors.newCachedThreadPool(new ThreadFactory(){
    int threadNumber=0;
    public Thread newThread(    Runnable r){
      threadNumber+=1;
      Thread t=new Thread(r,"LongRunning-" + threadNumber);
      t.setDaemon(true);
      return t;
    }
  }
);
  cronScheduler.setScheduledTaskExec(execService);
}
 

private void configureScheduledTasks(){
  if (policy.getTimeBetweenScheduledMaintenanceTaskRunsMillis() > 0) {
    if (policy.isRunMaintenanceTaskDuringInit()) {
      logger.info("Running maintenance tasks during initialization...");
      runMaintenanceTasks();
    }
    if (Policy.MIN_TIME_BETWEEN_SCHEDULED_TASKS >= policy.getTimeBetweenScheduledMaintenanceTaskRunsMillis()) {
      logger.warn("Setting the scheduled tasks to run less than every {} milliseconds is not a good idea...",Policy.MIN_TIME_BETWEEN_SCHEDULED_TASKS);
    }
    logger.info("Configuring scheduled tasks to run every {} milliseconds",policy.getTimeBetweenScheduledMaintenanceTaskRunsMillis());
    executorService=Executors.newScheduledThreadPool(1,new ThreadFactory(){
      @Override public Thread newThread(      Runnable runnable){
        Thread thread=new Thread(runnable,"pelops-pool-worker-" + getKeyspace());
        thread.setDaemon(true);
        thread.setPriority(Thread.MIN_PRIORITY + 1);
        return thread;
      }
    }
);
    executorService.scheduleWithFixedDelay(new Runnable(){
      @Override public void run(){
        logger.debug("Background thread running maintenance tasks");
        try {
          runMaintenanceTasks();
        }
 catch (        Exception e) {
          logger.warn("An exception was thrown while running the maintenance tasks",e);
        }
      }
    }
,policy.getTimeBetweenScheduledMaintenanceTaskRunsMillis(),policy.getTimeBetweenScheduledMaintenanceTaskRunsMillis(),TimeUnit.MILLISECONDS);
  }
 else {
    logger.warn("Disabling maintenance tasks; dynamic node discovery, node suspension, idle connection " + "termination and some running statistics will not be available to this pool.");
  }
}
 

/** 
 * Initializes this connector.
 * @throws ConnectorException if the initialization failed.
 */
protected final void initialize() throws ConnectorException {
synchronized (_isInitializedMutex) {
    if (!_isInitialized) {
      _asyncSender=Executors.newCachedThreadPool(new ThreadFactory(){
        private final AtomicInteger threadNumber=new AtomicInteger();
        public Thread newThread(        Runnable r){
          Thread thread=new Thread(r,"AsyncSkypeMessageSender-" + threadNumber.getAndIncrement());
          thread.setDaemon(true);
          return thread;
        }
      }
);
      _syncSender=Executors.newSingleThreadExecutor(new ThreadFactory(){
        public Thread newThread(        Runnable r){
          Thread thread=new Thread(r,"SyncSkypeMessageSender");
          thread.setDaemon(true);
          return thread;
        }
      }
);
      _commandExecutor=Executors.newCachedThreadPool(new ThreadFactory(){
        private final AtomicInteger threadNumber=new AtomicInteger();
        public Thread newThread(        Runnable r){
          Thread thread=new Thread(r,"CommandExecutor-" + threadNumber.getAndIncrement());
          thread.setDaemon(true);
          return thread;
        }
      }
);
      initializeImpl();
      _isInitialized=true;
    }
  }
}
 

/** 
 * Shuts down and cleans up all the Linux monitors mananged by this class.
 */
public void stopPlatformSpecificMonitoring(){
  ExecutorService executor=Executors.newFixedThreadPool(monitors.size(),new ThreadFactory(){
    public Thread newThread(    Runnable r){
      Thread t=new Thread(r);
      t.setName("Shutdown thread");
      return t;
    }
  }
);
  for (  Monitor s : monitors) {
    executor.submit(new ShutdownTask(s));
  }
  executor.shutdown();
  try {
    executor.awaitTermination(5,TimeUnit.SECONDS);
  }
 catch (  InterruptedException e) {
    System.out.println("Skipping orderly shutdown due to interrupt.");
  }
}
 

/** 
 * Creates a new instance of a OrderedThreadPoolExecutor.
 * @param corePoolSize The initial pool sizePoolSize
 * @param maximumPoolSize The maximum pool size
 * @param keepAliveTime Default duration for a thread
 * @param unit Time unit used for the keepAlive value
 * @param threadFactory The factory used to create threads
 * @param eventQueueHandler The queue used to store events
 */
public OrderedThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,ThreadFactory threadFactory,IoEventQueueHandler eventQueueHandler){
  super(DEFAULT_INITIAL_THREAD_POOL_SIZE,1,keepAliveTime,unit,new SynchronousQueue<Runnable>(),threadFactory,new AbortPolicy());
  if (corePoolSize < DEFAULT_INITIAL_THREAD_POOL_SIZE) {
    throw new IllegalArgumentException("corePoolSize: " + corePoolSize);
  }
  if ((maximumPoolSize == 0) || (maximumPoolSize < corePoolSize)) {
    throw new IllegalArgumentException("maximumPoolSize: " + maximumPoolSize);
  }
  super.setCorePoolSize(corePoolSize);
  super.setMaximumPoolSize(maximumPoolSize);
  if (eventQueueHandler == null) {
    this.eventQueueHandler=IoEventQueueHandler.NOOP;
  }
 else {
    this.eventQueueHandler=eventQueueHandler;
  }
}
 

public ConnectionNumberListener(int connThreshold,int ipCountThreshold,int ipCheckTaskInterval){
  this.connThreshold=connThreshold;
  this.ipCountThreshold=ipCountThreshold;
  this.ipCheckTaskInterval=ipCheckTaskInterval;
  this.scheduler=Executors.newSingleThreadScheduledExecutor(new ThreadFactory(){
    public Thread newThread(    Runnable r){
      Thread t=new Thread(r);
      t.setName("connection num control thread");
      t.setDaemon(true);
      return t;
    }
  }
);
  this.scheduler.scheduleAtFixedRate(new Runnable(){
    public void run(){
      int ipCount=ConnectionNumberListener.this.connectionIpNumMap.size();
      if (ipCount >= ConnectionNumberListener.this.ipCountThreshold) {
        log.warn("IP??????, ??????IP???, ??IP?=" + ipCount + ", ??="+ ConnectionNumberListener.this.ipCountThreshold);
        isOverflow=true;
      }
 else {
        isOverflow=false;
      }
    }
  }
,this.ipCheckTaskInterval,this.ipCheckTaskInterval,TimeUnit.SECONDS);
}
 

protected synchronized ExecutorService getService(){
  if (_executorService == null) {
    _executorService=Executors.newFixedThreadPool(_threadPoolSize,new ThreadFactory(){
      private int count=0;
      public Thread newThread(      Runnable runnable){
        Thread thread=new Thread(runnable,"image-pool-" + count++);
        thread.setPriority(Thread.MAX_PRIORITY);
        thread.setDaemon(true);
        return (thread);
      }
    }
);
  }
  return (_executorService);
}
 

@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;
}
 

public JMSTopicsContextImpl(ThreadFactory threadFactory,InitialContext initialContext,Name connectionFactoryName,Name topicsNamespace,String username,String password){
  super(threadFactory);
  this.initialContext=initialContext;
  this.connectionFactoryName=connectionFactoryName;
  this.topicsNamespace=topicsNamespace;
  this.username=username;
  this.password=password;
}
 

/** 
 * Sets the DaapThreadFactory for this DAAP server
 * @param fectory a DaapThreadFactory
 */
public synchronized void setThreadFactory(ThreadFactory factory){
  if (factory == null) {
    threadFactory=new DaapThreadFactory("DaapConnectionThread");
  }
 else {
    threadFactory=factory;
  }
}
 

public CancellableInputStream(InputStream stream,IProgressMonitor monitor,ThreadFactory factory){
  Utils.assertNotNullArg(stream,"inputStream");
  this.inputStream=stream;
  this.streamSupport=new CancellableStreamSupport(monitor,factory);
  this.readTask=new ReadCall(inputStream);
}
 

/** 
 * Creates a new timer.
 * @param threadFactory a  {@link ThreadFactory} that creates a background{@link Thread} which is dedicated to {@link TimerTask}execution.
 * @param tickDuration the duration between tick
 * @param unit the time unit of the  {@code tickDuration}
 * @param ticksPerWheel
 * @param maxTimerCapacity the size of the wheel
 */
public HashedWheelTimer(final ThreadFactory threadFactory,long tickDuration,final TimeUnit unit,final int ticksPerWheel,final int maxTimerCapacity){
  if (threadFactory == null) {
    throw new NullPointerException("threadFactory");
  }
  if (unit == null) {
    throw new NullPointerException("unit");
  }
  if (tickDuration <= 0) {
    throw new IllegalArgumentException("tickDuration must be greater than 0: " + tickDuration);
  }
  if (ticksPerWheel <= 0) {
    throw new IllegalArgumentException("ticksPerWheel must be greater than 0: " + ticksPerWheel);
  }
  if (maxTimerCapacity <= 0) {
    throw new IllegalArgumentException("maxTimerCapacity must be greater than 0: " + maxTimerCapacity);
  }
  this.wheel=createWheel(ticksPerWheel);
  this.iterators=createIterators(this.wheel);
  this.maxTimerCapacity=maxTimerCapacity;
  this.mask=this.wheel.length - 1;
  this.tickDuration=tickDuration=unit.toMillis(tickDuration);
  if (tickDuration == Long.MAX_VALUE || tickDuration >= Long.MAX_VALUE / this.wheel.length) {
    throw new IllegalArgumentException("tickDuration is too long: " + tickDuration + ' '+ unit);
  }
  this.roundDuration=tickDuration * this.wheel.length;
  this.workerThread=threadFactory.newThread(new ThreadRenamingRunnable(this.worker,"Hashed wheel timer #" + id.incrementAndGet()));
  final int activeInstances=HashedWheelTimer.activeInstances.incrementAndGet();
  if (activeInstances >= MISUSE_WARNING_THRESHOLD && loggedMisuseWarning.compareAndSet(false,true)) {
    logger.debug("There are too many active " + HashedWheelTimer.class.getSimpleName() + " instances ("+ activeInstances+ ") - you should share the small number "+ "of instances to avoid excessive resource consumption.");
  }
}
 

/** 
 * 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);
      }
    }
  }
);
}
 

/** 
 * Creates an instance of AbstractMultiworkerIOReactor with the given configuration.
 * @param config I/O reactor configuration.
 * @param threadFactory the factory to create threads.Can be <code>null</code>.
 * @throws IOReactorException in case if a non-recoverable I/O error.
 * @since 4.2
 */
public AbstractMultiworkerIOReactor(final IOReactorConfig config,final ThreadFactory threadFactory) throws IOReactorException {
  super();
  this.config=config != null ? config : IOReactorConfig.DEFAULT;
  this.params=new BasicHttpParams();
  try {
    this.selector=Selector.open();
  }
 catch (  IOException ex) {
    throw new IOReactorException("Failure opening selector",ex);
  }
  this.selectTimeout=this.config.getSelectInterval();
  this.interestOpsQueueing=this.config.isInterestOpQueued();
  this.statusLock=new Object();
  if (threadFactory != null) {
    this.threadFactory=threadFactory;
  }
 else {
    this.threadFactory=new DefaultThreadFactory();
  }
  this.workerCount=this.config.getIoThreadCount();
  this.dispatchers=new BaseIOReactor[workerCount];
  this.workers=new Worker[workerCount];
  this.threads=new Thread[workerCount];
  this.status=IOReactorStatus.INACTIVE;
}
 

/** 
 * create a scheduler
 * @param numThreads
 * @param baseThreadName
 * @param isDaemon
 */
public Scheduler(int numThreads,final String baseThreadName,final boolean isDaemon){
  this.baseThreadName=baseThreadName;
  executor=new ScheduledThreadPoolExecutor(numThreads,new ThreadFactory(){
    public Thread newThread(    Runnable r){
      Thread t=new Thread(r,baseThreadName + threadId.getAndIncrement());
      t.setDaemon(isDaemon);
      return t;
    }
  }
);
  executor.setContinueExistingPeriodicTasksAfterShutdownPolicy(false);
  executor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
}
 

/** 
 * Construct a new instance.
 * @param queueLength the queue length
 * @param threadFactory the thread factory to use to construct the handler thread
 */
public AsyncHandler(final int queueLength,final ThreadFactory threadFactory){
  recordQueue=new ArrayBlockingQueue<ExtLogRecord>(queueLength);
  thread=threadFactory.newThread(new AsyncTask());
  if (thread == null) {
    throw new IllegalArgumentException("Thread factory did not create a thread");
  }
  thread.setDaemon(true);
  this.queueLength=queueLength;
}
 

/** 
 * Create a WorkerPool with the given number of Workers and the given <code>ThreadFactory</code>.
 * @param workerFactory a Callable that returns an implementation of Worker
 * @param numWorkers the number of Workers to create
 * @param threadFactory the factory to create pre-configured Threads
 */
public WorkerPool(final Callable<? extends Worker> workerFactory,final int numWorkers,final ThreadFactory threadFactory){
  this.workers=new ArrayList<Worker>(numWorkers);
  this.threads=new ArrayList<Thread>(numWorkers);
  for (int i=0; i < numWorkers; i++) {
    try {
      final Worker worker=workerFactory.call();
      this.workers.add(worker);
      this.threads.add(threadFactory.newThread(worker));
    }
 catch (    RuntimeException re) {
      throw re;
    }
catch (    Exception e) {
      throw new RuntimeException(e);
    }
  }
}
 

/** 
 * Creates a new instance of a OrderedThreadPoolExecutor.
 * @param corePoolSize The initial pool sizePoolSize
 * @param maximumPoolSize The maximum pool size
 * @param keepAliveTime Default duration for a thread
 * @param unit Time unit used for the keepAlive value
 * @param threadFactory The factory used to create threads
 * @param eventQueueHandler The queue used to store events
 */
public OrderedThreadPoolExecutor(final int corePoolSize,final int maximumPoolSize,final long keepAliveTime,final TimeUnit unit,final ThreadFactory threadFactory){
  super(DEFAULT_INITIAL_THREAD_POOL_SIZE,1,keepAliveTime,unit,new SynchronousQueue<Runnable>(),threadFactory,new AbortPolicy());
  if (corePoolSize < DEFAULT_INITIAL_THREAD_POOL_SIZE) {
    throw new IllegalArgumentException("corePoolSize: " + corePoolSize);
  }
  if (maximumPoolSize == 0 || maximumPoolSize < corePoolSize) {
    throw new IllegalArgumentException("maximumPoolSize: " + maximumPoolSize);
  }
  super.setCorePoolSize(corePoolSize);
  super.setMaximumPoolSize(maximumPoolSize);
}
 

/** 
 * Creates a new timer.
 * @param threadFactory  a {@link ThreadFactory} that creates abackground  {@link Thread} which is dedicated to{@link TimerTask} execution.
 * @param tickDuration   the duration between tick
 * @param unit           the time unit of the {@code tickDuration}
 * @param ticksPerWheel  the size of the wheel
 */
public HashedWheelTimer(ThreadFactory threadFactory,long tickDuration,TimeUnit unit,int ticksPerWheel){
  if (threadFactory == null) {
    throw new NullPointerException("threadFactory");
  }
  if (unit == null) {
    throw new NullPointerException("unit");
  }
  if (tickDuration <= 0) {
    throw new IllegalArgumentException("tickDuration must be greater than 0: " + tickDuration);
  }
  if (ticksPerWheel <= 0) {
    throw new IllegalArgumentException("ticksPerWheel must be greater than 0: " + ticksPerWheel);
  }
  wheel=createWheel(ticksPerWheel);
  mask=wheel.length - 1;
  this.tickDuration=tickDuration=unit.toMillis(tickDuration);
  if (tickDuration == Long.MAX_VALUE || tickDuration >= Long.MAX_VALUE / wheel.length) {
    throw new IllegalArgumentException("tickDuration is too long: " + tickDuration + ' '+ unit);
  }
  roundDuration=tickDuration * wheel.length;
  workerThread=threadFactory.newThread(worker);
  misuseDetector.increase();
}
 

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 ScheduledExecutorService initClusterTaskScheduler() throws PlatformException {
  taskScheduler=new ScheduledThreadPoolExecutor(2,new ThreadFactory(){
    private final ThreadFactory factory=java.util.concurrent.Executors.defaultThreadFactory();
    @Override public Thread newThread(    Runnable r){
      Thread t=factory.newThread(r);
      t.setName("ODE-X Cluster Action Scheduler");
      t.setDaemon(true);
      return t;
    }
  }
);
  return taskScheduler;
}
 

public TcpServer(PascRuntime<PaxosState> runtime,StateMachine sm,ControlHandler controlHandler,String zk,String servers[],int port,int threads,final int id,boolean twoStages) throws IOException, KeeperException {
  this.bossExecutor=Executors.newCachedThreadPool();
  this.workerExecutor=Executors.newCachedThreadPool();
  this.executionHandler=new ExecutionHandler(new MemoryAwareThreadPoolExecutor(1,1024 * 1024,1024 * 1024 * 1024,10,TimeUnit.SECONDS,new ObjectSizeEstimator(){
    @Override public int estimateSize(    Object o){
      return 1024;
    }
  }
,new ThreadFactory(){
    private int count=0;
    @Override public Thread newThread(    Runnable r){
      return new Thread(r,id + "-" + count++);
    }
  }
));
  this.channelHandler=new ServerHandler(runtime,sm,controlHandler,this);
  this.channelPipelineFactory=new PipelineFactory(channelHandler,executionHandler,twoStages);
  this.leaderElection=new LeaderElection(zk,id,this.channelHandler);
  this.barrier=new Barrier(new ZooKeeper(zk,2000,leaderElection),"/paxos_srv_barrier","" + id,servers.length);
  this.servers=servers;
  this.port=port;
  this.threads=threads;
  this.id=id;
}
 

protected ExecutorService initializeExecutor(ThreadFactory threadFactory,RejectedExecutionHandler rejectedExecutionHandler){
  ScheduledExecutorService executor=createExecutor(this.poolSize,threadFactory,rejectedExecutionHandler);
  if (!ObjectUtils.isEmpty(this.scheduledExecutorTasks)) {
    registerTasks(this.scheduledExecutorTasks,executor);
  }
  this.exposedExecutor=(this.exposeUnconfigurableExecutor ? Executors.unconfigurableScheduledExecutorService(executor) : executor);
  return executor;
}
 

/** 
 * Constructs a new abstract service with the given name.
 * @param name The name of the service
 * @param registerShutdownHook <code>true</code> to register shutdown hook for this service, <code>false</code> to not register a shutdown hook
 * @param threadFactory The thread factory used to create the service thread
 */
protected AbstractService(String name,boolean registerShutdownHook,ThreadFactory threadFactory){
  this.registerShutdownHook=registerShutdownHook;
  Validation.begin().isNotNull("name",name).isNotNull("threadFactory",threadFactory).check();
  this.name=name;
  this.threadFactory=threadFactory;
}
 

/** 
 * 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());
  }
}
 

/** 
 * 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());
}