Java Code Examples for java.util.concurrent.ArrayBlockingQueue

/** 
 * Perform a roster sync on a given account and a given content provider.
 * @param account The xmpp account to be synced.
 * @param extras SyncAdapter-specific parameters
 * @param authority The authority of this sync request.
 * @param provider A authority based ContentProvider for this sync.
 * @param syncResult Sync error and result counters.
 */
@Override public void onPerformSync(final Account account,Bundle extras,String authority,ContentProviderClient provider,SyncResult syncResult){
  Log.d(TAG,"Start Roster Sync");
  final ArrayBlockingQueue<Node> rosterQueue=new ArrayBlockingQueue<Node>(1);
  BroadcastReceiver receiver=new RosterResultReceiver(account,rosterQueue);
  try {
    bindService();
    if (!waitForService()) {
      return;
    }
    if (!waitForServiceBind(account.name)) {
      return;
    }
    Stanza stanza=getRosterRequest(account);
    if (!sendWithRetry(stanza)) {
      syncResult.stats.numIoExceptions++;
      return;
    }
    Node roster=rosterQueue.poll(300,TimeUnit.SECONDS);
    if (roster == null) {
      return;
    }
    handleRosterResult(account,roster,provider);
  }
 catch (  InterruptedException e) {
    Log.e(TAG,"Sync interrupted",e);
  }
 finally {
    applicationContext.unregisterReceiver(receiver);
    unbindService();
  }
}
 

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

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

/** 
 * Perform a roster sync on a given account and a given content provider.
 * @param account The xmpp account to be synced.
 * @param extras SyncAdapter-specific parameters
 * @param authority The authority of this sync request.
 * @param provider A authority based ContentProvider for this sync.
 * @param syncResult Sync error and result counters.
 */
@Override public void onPerformSync(final Account account,Bundle extras,String authority,ContentProviderClient provider,SyncResult syncResult){
  Log.d(TAG,"Start Roster Sync");
  final ArrayBlockingQueue<Node> rosterQueue=new ArrayBlockingQueue<Node>(1);
  BroadcastReceiver receiver=new RosterResultReceiver(account,rosterQueue);
  applicationContext.registerReceiver(receiver,new IntentFilter(XmppTransportService.XMPP_STANZA_INTENT));
  try {
    bindService();
    if (!waitForService()) {
      return;
    }
    if (!waitForServiceBind(account.name)) {
      return;
    }
    Stanza stanza=getRosterRequest(account);
    if (!sendWithRetry(stanza)) {
      syncResult.stats.numIoExceptions++;
      return;
    }
    Node roster=rosterQueue.poll(300,TimeUnit.SECONDS);
    if (roster == null) {
      return;
    }
    handleRosterResult(account,roster,provider);
  }
 catch (  InterruptedException e) {
    Log.e(TAG,"Sync interrupted",e);
  }
 finally {
    applicationContext.unregisterReceiver(receiver);
    unbindService();
  }
}
 

/** 
 * If we hit our limit, we should signal for more connections to be created on the fly
 * @throws SQLException
 */
@Test public void testGetConnectionLimitsHit() throws SQLException {
  reset(mockPartition,mockConnectionHandles,mockConnection);
  expect(mockConfig.getPoolAvailabilityThreshold()).andReturn(0).anyTimes();
  expect(mockPartition.isUnableToCreateMoreTransactions()).andReturn(false).anyTimes();
  expect(mockPartition.getFreeConnections()).andReturn(mockConnectionHandles).anyTimes();
  expect(mockPartition.getMaxConnections()).andReturn(10).anyTimes();
  expect(mockPartition.getAvailableConnections()).andReturn(1).anyTimes();
  BlockingQueue<Object> bq=new ArrayBlockingQueue<Object>(1);
  bq.add(new Object());
  expect(mockPartition.getPoolWatchThreadSignalQueue()).andReturn(bq);
  expect(mockConnectionHandles.poll()).andReturn(mockConnection).once();
  mockConnection.renewConnection();
  expectLastCall().once();
  replay(mockPartition,mockConnectionHandles,mockConnection);
  testClass.getConnection();
  verify(mockPartition,mockConnectionHandles,mockConnection);
}
 

/** 
 * <p> Singleton constructor </p> <p> If there is an exception, we log the error and exit because there's no point continuing without MQ client session </p>
 * @param poolSize number of session to pool
 * @throws Exception
 */
public MqProducerSessionPool(int poolSize){
  try {
    _pool=new ArrayBlockingQueue<Pooled>(poolSize);
    for (int i=0; i < poolSize; i++) {
      addPooled();
    }
    return;
  }
 catch (  Exception e) {
    _logger.error("Error loading Publisher Session Pool: " + e.getMessage(),e);
    System.exit(1);
  }
}
 

public Tee(Socket s) throws IOException {
  sock=s;
  sendQ=new ArrayBlockingQueue<Chunk>(QUEUE_LENGTH);
  Thread t=new Thread(this);
  t.setDaemon(true);
  t.start();
}
 

public CompilationOverseer(final CompilationSettings settings,final Log log,final Collection<NativeCodeFile> allClasses,final AbstractCompiler compiler){
  this.settings=settings;
  this.log=log;
  this.allCodeFiles=allClasses;
  this.compiler=compiler;
  this.classesToCompile=new ArrayBlockingQueue<NativeCodeFile>(allClasses.size());
  this.compiledClasses=new ConcurrentLinkedQueue<NativeCodeFile>();
  final int numberOfCompilerThreads=getNumberOfCompilerThreads();
  for (int i=0; i < numberOfCompilerThreads; i++)   processors.add(new CompilationProcessor("Compilation Processor " + i,compiler,log,classesToCompile,compiledClasses,numberOfCompilerThreads,monitor));
}
 

public PooledJaxb(JAXBContext jaxbContext,int poolSize,long maxWait) throws JAXBException {
  this.maxWait=maxWait;
  this.pool=new ArrayBlockingQueue<Jaxb>(poolSize);
  for (int i=0; i < poolSize; i++) {
    SimpleJaxb jaxb=new SimpleJaxb(jaxbContext);
    this.pool.add(jaxb);
  }
}
 

public Emitter(Writer stream,DumperOptions opts){
  this.stream=stream;
  this.states=new ArrayStack<EmitterState>(100);
  this.state=new ExpectStreamStart();
  this.events=new ArrayBlockingQueue<Event>(100);
  this.event=null;
  this.indents=new ArrayStack<Integer>(10);
  this.indent=null;
  this.flowLevel=0;
  mappingContext=false;
  simpleKeyContext=false;
  column=0;
  whitespace=true;
  indention=true;
  openEnded=false;
  this.canonical=opts.isCanonical();
  this.prettyFlow=opts.isPrettyFlow();
  this.allowUnicode=opts.isAllowUnicode();
  this.bestIndent=2;
  if ((opts.getIndent() > MIN_INDENT) && (opts.getIndent() < MAX_INDENT)) {
    this.bestIndent=opts.getIndent();
  }
  this.bestWidth=80;
  if (opts.getWidth() > this.bestIndent * 2) {
    this.bestWidth=opts.getWidth();
  }
  this.bestLineBreak=opts.getLineBreak().getString().toCharArray();
  this.tagPrefixes=new LinkedHashMap<String,String>();
  this.preparedAnchor=null;
  this.preparedTag=null;
  this.analysis=null;
  this.style=null;
  this.options=opts;
}
 

/** 
 * Set the maxRequestsNumber property. 
 */
public void setMaxRequestsNumber(final String maxRequestsNumber){
  log.info("[AUTOREARMFUSE INTENT] : setting maxRequestsNumber to '" + maxRequestsNumber + "'.");
  try {
    int mrn=Integer.parseInt(maxRequestsNumber);
    if (mrn > 0) {
      this.maxRequestsNumber=mrn;
      requestQueue=new ArrayBlockingQueue<RequestElement>(this.maxRequestsNumber);
    }
  }
 catch (  Exception ex) {
    log.error("[AUTOREARMFUSE INTENT] : Invalid value for maxRequestsNumber property. Default value will be use instead !");
  }
}
 

public ChannelPipeline getPipeline() throws Exception {
  ChannelPipeline pipeline=Channels.pipeline();
  addLengthFieldPipes(pipeline,frameLength);
  BlockingReadHandler<ChannelBuffer> reader=new BlockingReadHandler<ChannelBuffer>(new ArrayBlockingQueue<ChannelEvent>(3,false));
  pipeline.addLast("blockingHandler",reader);
  return pipeline;
}
 

/** 
 * @param parser
 * @param path an XPath query
 * @return
 * @throws XmlPullParserException
 * @throws IOException
 */
public static boolean accessPath(XmlPullParser parser,String path) throws XmlPullParserException, IOException {
  String[] pathElements=path.split("/");
  Queue<String> pathQueue=new ArrayBlockingQueue<String>(pathElements.length);
  for (  String pathElement : pathElements) {
    pathQueue.offer(pathElement);
  }
  return accessPath(parser,pathQueue);
}
 

public BatchProcessor(String name,BatchHandler<T> handler,int maxBatchSize,int queueSize){
  Preconditions.checkNotNull(name,"name is null");
  Preconditions.checkNotNull(handler,"handler is null");
  Preconditions.checkArgument(queueSize > 0,"queue size needs to be a positive integer");
  Preconditions.checkArgument(maxBatchSize > 0,"max batch size needs to be a positive integer");
  this.handler=handler;
  this.maxBatchSize=maxBatchSize;
  this.queue=new ArrayBlockingQueue<T>(queueSize);
  this.executor=Executors.newSingleThreadExecutor(new ThreadFactoryBuilder().setNameFormat(format("batch-processor-%s",name)).build());
}
 

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

@Override public void configure(Context context){
  Integer capacity=context.getInteger("capacity");
  keepAlive=context.getInteger("keep-alive");
  if (capacity == null) {
    capacity=defaultCapacity;
  }
  if (keepAlive == null) {
    keepAlive=defaultKeepAlive;
  }
  queue=new ArrayBlockingQueue<Event>(capacity);
  if (channelCounter == null) {
    channelCounter=new ChannelCounter(getName());
  }
}
 

/** 
 * Pre-existing file, start cursor, and check we get # of events we expected
 */
@Test public void testCursorPrexisting() throws IOException, InterruptedException {
  BlockingQueue<Event> q=new ArrayBlockingQueue<Event>(100);
  File f=createDataFile(5);
  Cursor c=new Cursor(q,f);
  assertTrue(c.tailBody());
  assertTrue(c.tailBody());
  assertFalse(c.tailBody());
  assertEquals(5,q.size());
}
 

@Override public void open() throws IOException {
  if (null != mFlumeNodeName) {
    if (!mFlumeConfig.isRunning()) {
      mFlumeConfig.start();
    }
    mOutputQueue=new ArrayBlockingQueue<Event>(MAX_QUEUE_LEN);
    SourceContext srcContext=new SourceContext(mFlumeNodeName,mOutputQueue);
    SourceContextBindings.get().bindContext(mFlumeNodeName,srcContext);
    try {
      mFlumeConfig.spawnLogicalNode(mFlumeNodeName,"rtsqlsource(\"" + mFlumeNodeName + "\")","rtsqlmultisink(\"" + mFlumeNodeName + "\")");
      mFlumeConfig.addLocalMultiSink(mFlumeNodeName);
      if (mRootSymbolTable.resolve(mFlumeNodeName) != null) {
        LOG.error("Cannot create stream for flow; object already exists at top level: " + mFlumeNodeName);
        mOwnsSymbol=false;
        ((LocalContext)getContext()).getFlowData().setStreamName(null);
      }
 else {
        FormatSpec formatSpec=new FormatSpec(FormatSpec.FORMAT_AVRO);
        formatSpec.setParam(AvroEventParser.SCHEMA_PARAM,mOutputSchema.toString());
        List<Type> outputTypes=new ArrayList<Type>();
        for (        TypedField field : mFlumeInputFields) {
          outputTypes.add(field.getType());
        }
        Type streamType=new StreamType(outputTypes);
        StreamSymbol streamSym=new StreamSymbol(mFlumeNodeName,StreamSourceType.Node,streamType,mFlumeNodeName,true,mOutputFields,formatSpec);
        if (!streamSym.getEventParser().validate(streamSym)) {
          throw new IOException("Could not create valid stream for schema");
        }
        mRootSymbolTable.addSymbol(streamSym);
        mOwnsSymbol=true;
        ((LocalContext)getContext()).getFlowData().setStreamName(mFlumeNodeName);
        LOG.info("CREATE STREAM (" + mFlumeNodeName + ")");
      }
    }
 catch (    TException te) {
      throw new IOException(te);
    }
  }
}
 

/** 
 * This checks to make sure that tags are always get lexographically larger over time. A ProcessTagger actually uses thread id # as part of its sort and this verifies that it is the least significant. A Roller can call the new tag method in either of its threads, so we need to take this into account.
 */
@Test public void testThreadedTaggerNameMonotonic() throws InterruptedException {
  final Tagger t=new ProcessTagger();
  final Queue<String> tags=new ArrayBlockingQueue<String>(1000);
  final Object lock=new Object();
  final CountDownLatch start=new CountDownLatch(10);
  final CountDownLatch done=new CountDownLatch(10);
class TagThread extends Thread {
    public void run(){
      start.countDown();
      try {
        start.await();
        while (true) {
synchronized (lock) {
            String s=t.newTag();
            boolean accepted=tags.offer(s);
            if (!accepted) {
              done.countDown();
              return;
            }
            LOG.info("added tag: {}",s);
          }
        }
      }
 catch (      InterruptedException e) {
        e.printStackTrace();
      }
    }
  }
  TagThread[] thds=new TagThread[10];
  for (int i=0; i < thds.length; i++) {
    thds[i]=new TagThread();
    thds[i].start();
  }
  done.await();
  String[] aTags=tags.toArray(new String[0]);
  for (int i=1; i < aTags.length; i++) {
    assertTrue(aTags[i - 1].compareTo(aTags[i]) < 0);
  }
}
 

/** 
 * Override to use different thread or queue. Not recommended.
 */
protected static void initQueue(){
  if (queue == null) {
    queue=new ArrayBlockingQueue<QueuedCommandWidget>(3);
  }
  if (mainQueueThread == null) {
    mainQueueThread=uk.ac.gda.util.ThreadManager.getThread(getRunnable(),"QueuedCommandWidget thread. Used to updated all " + QueuedCommandWidget.class.getName() + "'s");
    mainQueueThread.start();
  }
}
 

@SuppressWarnings("unchecked") @Override public synchronized <E>BlockingQueue<E> queue(final String name){
  if (!queues.containsKey(name)) {
    queues.put(name,new ArrayBlockingQueue<E>(queueCapacity));
  }
  return (BlockingQueue<E>)queues.get(name);
}
 

@Inject public ASMClasspathScanner(Set<ScannerFeature> listeners,@Named("packages") PackageFilter... filter){
  int cores=Runtime.getRuntime().availableProcessors();
  this.collectors=new ArrayBlockingQueue<AnnotationCollector>(cores);
  for (int i=0; i < cores; i++) {
    try {
      collectors.put(new AnnotationCollector());
    }
 catch (    InterruptedException e) {
    }
  }
  for (  PackageFilter p : filter) {
    includePackage(p);
  }
  for (  ScannerFeature listener : listeners) {
    addFeature(listener);
  }
  visited=Collections.synchronizedSet(new HashSet<String>());
}
 

public BlockingPipe(int queueCapacity){
  super();
  queue=new ArrayBlockingQueue<Object>(queueCapacity);
  closeables.addFirst(input);
  closeables.addFirst(output);
}
 

private void init(){
  if (initialized)   return;
  tupleLoaders=new HashMap<TableDesc[],TupleLoader>();
  currentLoader=null;
  count=0;
  if (threading) {
    queue=new ArrayBlockingQueue<TupleChange>(chunkSize);
    threadException=new AtomicReference<Throwable>();
    threadFlushing=new AtomicBoolean();
    commitThread=new Thread(new Commiter());
    commitThread.setDaemon(true);
    commitThread.start();
    LOG.debug("Threading started");
  }
  initialized=true;
}
 

private static void putDFSClient(Configuration conf,DFSClient client) throws IOException {
  BlockingQueue<DFSClient> clientQueue;
synchronized (clients) {
    clientQueue=clients.get(conf);
    if (clientQueue == null) {
      clientQueue=new ArrayBlockingQueue<DFSClient>(1);
      clients.put(conf,clientQueue);
    }
  }
  if (!clientQueue.offer(client)) {
    client.close();
  }
}
 

/** 
 * Constructor
 * @param serial  Used to generate unique filename sequences
 * @param factory Factory that knows how to make a {@link WriterPoolMember}.
 * @param settings Settings for this pool.
 * @param poolMaximumActive
 * @param poolMaximumWait
 */
public WriterPool(final AtomicInteger serial,final WriterPoolSettings settings,final int poolMaximumActive,final int poolMaximumWait){
  logger.info("Initial configuration:" + " prefix=" + settings.getPrefix() + ", suffix="+ settings.getTemplate()+ ", compress="+ settings.getCompress()+ ", maxSize="+ settings.getMaxFileSizeBytes()+ ", maxActive="+ poolMaximumActive+ ", maxWait="+ poolMaximumWait);
  this.settings=settings;
  this.maxActive=poolMaximumActive;
  this.maxWait=poolMaximumWait;
  availableWriters=new ArrayBlockingQueue<WriterPoolMember>(LARGEST_MAX_ACTIVE,true);
  this.serialNo=serial;
}
 

public Tee(Socket s) throws IOException {
  sock=s;
  sendQ=new ArrayBlockingQueue<Chunk>(QUEUE_LENGTH);
  Thread t=new Thread(this);
  t.setDaemon(true);
  t.start();
}
 

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

/** 
 * Create an LSI and its components
 * @param scorer The scorer to use when ranking results
 * @param parser 
 * @param basePath The base path (a directory) from the which all the LSI directories will be found.   
 */
public LargeScaleIndex(Scorer scorer,IndexEngineParser parser,File baseDir,FacetingManager facetingManager){
  Preconditions.checkNotNull(scorer);
  Preconditions.checkNotNull(parser);
  Preconditions.checkNotNull(baseDir);
  this.baseDir=baseDir;
  if (!baseDir.exists() || !baseDir.isDirectory()) {
    throw new IllegalArgumentException("The basePath must be an existing directory");
  }
  File indexDir=new File(baseDir,INDEX_DIRECTORY);
  if (!indexDir.exists()) {
    logger.info("Starting with a FRESH, BRAND NEW index.");
    indexDir.mkdir();
  }
  try {
    index=new LsiIndex(parser,indexDir.getAbsolutePath(),scorer,facetingManager);
  }
 catch (  IOException e) {
    throw new IllegalArgumentException("IOException when trying to use the directory set in the index.directory property.",e);
  }
  this.scorer=scorer;
  this.indexer=new LsiIndexer(index);
  this.searcher=new LsiSearcher(index);
  this.queue=new ArrayBlockingQueue<Operation>(1000);
  this.rwl=new ReentrantReadWriteLock();
  this.r=rwl.readLock();
  this.w=rwl.writeLock();
  this.checkpoint=false;
}
 

/** 
 * creaet a new thread processor
 * @param requesthandlerFactory request handler factory
 * @param stats jmx state statics
 * @param maxRequestSize max request package size
 * @param maxCacheConnections max cache connections for self-protected
 */
public Processor(RequestHandlerFactory requesthandlerFactory,SocketServerStats stats,int maxRequestSize,int maxCacheConnections){
  this.requesthandlerFactory=requesthandlerFactory;
  this.stats=stats;
  this.maxRequestSize=maxRequestSize;
  this.newConnections=new ArrayBlockingQueue<SocketChannel>(maxCacheConnections);
}
 

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

/** 
 * Set the configured work queue.  If the service is already started, the change will take effect upon next restart.
 * @param workQueue the work queue
 */
public synchronized void setWorkQueue(final BlockingQueue<Runnable> workQueue){
  if (workQueue == null) {
    setWorkQueue(new ArrayBlockingQueue<Runnable>(DEFAULT_QUEUE_LENGTH));
  }
  this.workQueue=workQueue;
}
 

/** 
 * Evaluate the classifier on all elements of a set
 * @param l the list of sample to classify
 * @return a list containing evaluation of the samples
 */
private List<Evaluation<TrainingSample<T>>> evaluateSet(final List<TrainingSample<T>> l){
  final List<Evaluation<TrainingSample<T>>> results=new ArrayList<Evaluation<TrainingSample<T>>>();
  int nbcpu=Runtime.getRuntime().availableProcessors();
  int length=l.size();
  ThreadPoolExecutor threadPool=new ThreadPoolExecutor(nbcpu,nbcpu,10,TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(length + 2));
  for (int i=length - 1; i >= 0; i--) {
    final int index=i;
    Runnable r=new Runnable(){
      @Override public void run(){
        TrainingSample<T> s=l.get(index);
        double r=classifier.valueOf(s.sample);
        Evaluation<TrainingSample<T>> e=new Evaluation<TrainingSample<T>>(s,r);
synchronized (results) {
          results.add(e);
        }
      }
    }
;
    threadPool.execute(r);
  }
  threadPool.shutdown();
  try {
    threadPool.awaitTermination(Integer.MAX_VALUE,TimeUnit.DAYS);
  }
 catch (  InterruptedException e) {
    debug.println(1,"Evaluator error - result corrupted");
    e.printStackTrace();
  }
  return results;
}
 

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

public ProcessPoolOfficeManager(File officeHome,UnoUrl[] unoUrls,String[] runAsArgs,File templateProfileDir,File workDir,long retryTimeout,long taskQueueTimeout,long taskExecutionTimeout,int maxTasksPerProcess,ProcessManager processManager){
  this.taskQueueTimeout=taskQueueTimeout;
  pool=new ArrayBlockingQueue<PooledOfficeManager>(unoUrls.length);
  pooledManagers=new PooledOfficeManager[unoUrls.length];
  for (int i=0; i < unoUrls.length; i++) {
    PooledOfficeManagerSettings settings=new PooledOfficeManagerSettings(unoUrls[i]);
    settings.setRunAsArgs(runAsArgs);
    settings.setTemplateProfileDir(templateProfileDir);
    settings.setWorkDir(workDir);
    settings.setOfficeHome(officeHome);
    settings.setRetryTimeout(retryTimeout);
    settings.setTaskExecutionTimeout(taskExecutionTimeout);
    settings.setMaxTasksPerProcess(maxTasksPerProcess);
    settings.setProcessManager(processManager);
    pooledManagers[i]=new PooledOfficeManager(settings);
  }
  logger.info("ProcessManager implementation is " + processManager.getClass().getSimpleName());
}
 

public ProcessPoolOfficeManager(File officeHome,UnoUrl[] unoUrls,File templateProfileDir,long taskQueueTimeout,long taskExecutionTimeout,int maxTasksPerProcess,ProcessManager processManager,boolean useGnuStyleLongOptions){
  this.taskQueueTimeout=taskQueueTimeout;
  pool=new ArrayBlockingQueue<PooledOfficeManager>(unoUrls.length);
  pooledManagers=new PooledOfficeManager[unoUrls.length];
  for (int i=0; i < unoUrls.length; i++) {
    PooledOfficeManagerSettings settings=new PooledOfficeManagerSettings(unoUrls[i]);
    settings.setTemplateProfileDir(templateProfileDir);
    settings.setOfficeHome(officeHome);
    settings.setTaskExecutionTimeout(taskExecutionTimeout);
    settings.setMaxTasksPerProcess(maxTasksPerProcess);
    settings.setProcessManager(processManager);
    settings.setUseGnuStyleLongOptions(useGnuStyleLongOptions);
    pooledManagers[i]=new PooledOfficeManager(settings);
  }
  logger.info("ProcessManager implementation is " + processManager.getClass().getSimpleName());
}
 

public ConsoleImpl(CommandProcessor processor,InputStream in,PrintStream out,PrintStream err,Terminal term,String encoding,Runnable closeCallback){
  this.in=in;
  this.out=out;
  this.err=err;
  this.queue=new ArrayBlockingQueue<Integer>(1024);
  this.terminal=term == null ? new UnsupportedTerminal() : term;
  this.consoleInput=new ConsoleInputStream();
  this.session=processor.createSession(this.consoleInput,this.out,this.err);
  this.session.put("SCOPE","shell:bundle:*");
  this.session.put("SUBSHELL","");
  this.closeCallback=closeCallback;
  try {
    reader=new ConsoleReader(null,this.consoleInput,this.out,this.terminal,encoding);
  }
 catch (  IOException e) {
    throw new RuntimeException("Error opening console reader",e);
  }
  final File file=getHistoryFile();
  try {
    file.getParentFile().mkdirs();
    reader.setHistory(new KarafFileHistory(file));
  }
 catch (  Exception e) {
    LOGGER.error("Can not read history from file " + file + ". Using in memory history",e);
  }
  session.put(".jline.reader",reader);
  session.put(".jline.history",reader.getHistory());
  Completer completer=createCompleter();
  if (completer != null) {
    reader.addCompleter(new CompleterAsCompletor(completer));
  }
  pipe=new Thread(new Pipe());
  pipe.setName("gogo shell pipe thread");
  pipe.setDaemon(true);
}
 

public MultiplexSendBytesService(String name,List<String> remoteHostsList,int port,WriteByteStrategy writeByteStrategy,int reconnectionDelay,int queueSize){
  this.name=name;
  this.queueSize=queueSize;
  this.remoteHostsList=remoteHostsList;
  this.senderServices=new HashSet<SimpleSendBytesService>();
  this.eventBytes=new ArrayBlockingQueue<byte[]>(queueSize,true);
  this.writeByteStrategy=writeByteStrategy;
  this.port=port;
  this.reconnectionDelay=reconnectionDelay;
}
 

@Override public void start(){
  if (appenderCount == 0) {
    addError("No attached appenders found.");
    return;
  }
  if (queueSize < 1) {
    addError("Invalid queue size [" + queueSize + "]");
    return;
  }
  blockingQueue=new ArrayBlockingQueue<E>(queueSize);
  if (discardingThreshold == UNDEFINED)   discardingThreshold=queueSize / 5;
  addInfo("Setting discardingThreshold to " + discardingThreshold);
  worker.setDaemon(true);
  worker.setName("AsyncAppender-Worker-" + worker.getName());
  super.start();
  worker.start();
}
 

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

public AgentClient(String authSocket) throws IOException {
  try {
    this.authSocket=authSocket;
    pool=Pool.create(AprLibrary.getInstance().getRootPool());
    handle=Local.create(authSocket,pool);
    int result=Local.connect(handle,0);
    if (result != Status.APR_SUCCESS) {
      throwException(result);
    }
    receiveBuffer=new Buffer();
    messages=new ArrayBlockingQueue<Buffer>(10);
    new Thread(this).start();
  }
 catch (  IOException e) {
    throw e;
  }
catch (  Exception e) {
    throw new SshException(e);
  }
}
 

public RemoteSniffingDaemon(IProbe probe,ILogger logger,IEventHandler changeHandler){
  this.probe=probe;
  this.logger=logger;
  stateChangeListeners=new EventListenerManager();
  stateChangeListeners.addListener(changeHandler);
  sendQueue=new ArrayBlockingQueue<IProbeMessage>(10);
  sendThread=new Thread(createSendMessageRunnable());
  sendThread.start();
  refreshInterfaceInformation();
  refreshModuleInformation();
  refreshStatus();
}
 

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 TSOClient(Configuration conf) throws IOException {
  state=State.DISCONNECTED;
  queuedOps=new ArrayBlockingQueue<Op>(200);
  retryTimer=new Timer(true);
  commitCallbacks=Collections.synchronizedMap(new HashMap<Long,CommitCallback>());
  isCommittedCallbacks=Collections.synchronizedMap(new HashMap<Long,List<CommitQueryCallback>>());
  createCallbacks=new ConcurrentLinkedQueue<CreateCallback>();
  channel=null;
  System.out.println("Starting TSOClient");
  factory=new NioClientSocketChannelFactory(Executors.newCachedThreadPool(),Executors.newCachedThreadPool(),3);
  bootstrap=new ClientBootstrap(factory);
  int executorThreads=conf.getInt("tso.executor.threads",3);
  bootstrap.getPipeline().addLast("executor",new ExecutionHandler(new OrderedMemoryAwareThreadPoolExecutor(executorThreads,1024 * 1024,4 * 1024 * 1024)));
  bootstrap.getPipeline().addLast("handler",this);
  bootstrap.setOption("tcpNoDelay",false);
  bootstrap.setOption("keepAlive",true);
  bootstrap.setOption("reuseAddress",true);
  bootstrap.setOption("connectTimeoutMillis",100);
  String host=conf.get("tso.host");
  int port=conf.getInt("tso.port",1234);
  max_retries=conf.getInt("tso.max_retries",100);
  retry_delay_ms=conf.getInt("tso.retry_delay_ms",1000);
  if (host == null) {
    throw new IOException("tso.host missing from configuration");
  }
  addr=new InetSocketAddress(host,port);
  connectIfNeeded();
}
 

/** 
 * 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 <L extends Mergeable<L>>Execution<L> submit(Computation<L> computation){
  int numberOfInstances=computation.getClass().getAnnotation(NumberOfInstances.class) == null ? getConfiguration().getNumberOfThreadsInSharedMemory() : computation.getClass().getAnnotation(NumberOfInstances.class).value();
  Collection<ComputationId> ids=new ArrayList<>(numberOfInstances);
  AtomicInteger maxId=new AtomicInteger(numberOfInstances - 1);
  for (int i=0; i < numberOfInstances; i++) {
    ids.add(new SharedMemoryComputationId(i,maxId));
  }
  ComputationContext context=new ComputationContext();
  getComputationContextEvent().initialize(context);
  BlockingQueue<Future<L>> futures=new ArrayBlockingQueue<>(getConfiguration().getQueueSize());
  context.getStorage().add(ComputationEmitter.class,Default.class,new SharedMemoryComputationEmitter<>(runnableExecutor,getEnrichment(),getComputationInstanceContextEvent(),context,maxId,futures));
  executeMethodsByAnnotation(getEnrichment(),context,computation,Before.class);
  return new SharedMemoryExecution<>(ids,runnableExecutor,computation,getEnrichment(),getComputationInstanceContextEvent(),context,futures);
}
 

/** 
 * The start method should only be called once, before the append method is called, to initialize options.
 * @param handler The GroupedTimingStatisticsHandler used to process GroupedTimingStatistics created by aggregatingStopWatch log message.
 */
public void start(GroupedTimingStatisticsHandler handler){
  if (drainingThread != null) {
    stopDrainingThread();
  }
  this.handler=handler;
  stopWatchParser=newStopWatchParser();
  numDiscardedMessages=0;
  loggedMessages=new ArrayBlockingQueue<String>(getQueueSize());
  drainingThread=new Thread(new Dispatcher(),"perf4j-async-stats-appender-sink-" + getName());
  drainingThread.setDaemon(true);
  drainingThread.start();
}
 

public void testConsumerCancellationNotification() throws IOException, InterruptedException {
  final BlockingQueue<Boolean> result=new ArrayBlockingQueue<Boolean>(1);
  channel.queueDeclare(queue,false,true,false,null);
  Consumer consumer=new QueueingConsumer(channel){
    @Override public void handleCancel(    String consumerTag) throws IOException {
      try {
        result.put(true);
      }
 catch (      InterruptedException e) {
        fail();
      }
    }
  }
;
  channel.basicConsume(queue,consumer);
  channel.queueDelete(queue);
  assertTrue(result.take());
}
 

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

private BlockingQueue<Runnable> threadQueue(int size){
switch (size) {
case -1:
    return new LinkedBlockingQueue<Runnable>();
case 0:
  return new SynchronousQueue<Runnable>();
default :
return new ArrayBlockingQueue<Runnable>(size);
}
}
 

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

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

Collection<Queue<Boolean>> queues(){
  List<Queue<Boolean>> queues=new ArrayList<Queue<Boolean>>();
  queues.add(new ConcurrentLinkedQueue<Boolean>());
  queues.add(new ArrayBlockingQueue<Boolean>(count,false));
  queues.add(new ArrayBlockingQueue<Boolean>(count,true));
  queues.add(new LinkedBlockingQueue<Boolean>());
  queues.add(new LinkedBlockingDeque<Boolean>());
  queues.add(new PriorityBlockingQueue<Boolean>());
  queues.add(new PriorityQueue<Boolean>());
  queues.add(new LinkedList<Boolean>());
  Collections.shuffle(queues);
  return queues;
}
 

private static <T>Collection<T> buildReplacementCollection(Collection<T> in) throws Throwable {
  int size=in.size();
  if (in.getClass().isInterface()) {
    if (in.getClass().equals(Set.class)) {
      return new HashSet<T>(size);
    }
    if (in.getClass().equals(List.class)) {
      return new ArrayList<T>(size);
    }
    if (in.getClass().equals(Queue.class)) {
      return new ArrayBlockingQueue<T>(size);
    }
  }
 else {
    return in.getClass().newInstance();
  }
  throw new RuntimeException("we couldn't figure out a replacement collection for the input collection type, " + in.getClass().getName());
}
 

protected Message doSendAndReceiveWithTemporary(final String exchange,final String routingKey,final Message message){
  Message replyMessage=this.execute(new ChannelCallback<Message>(){
    public Message doInRabbit(    Channel channel) throws Exception {
      final ArrayBlockingQueue<Message> replyHandoff=new ArrayBlockingQueue<Message>(1);
      Assert.isNull(message.getMessageProperties().getReplyTo(),"Send-and-receive methods can only be used if the Message does not already have a replyTo property.");
      DeclareOk queueDeclaration=channel.queueDeclare();
      String replyTo=queueDeclaration.getQueue();
      message.getMessageProperties().setReplyTo(replyTo);
      boolean noAck=true;
      String consumerTag=UUID.randomUUID().toString();
      boolean noLocal=true;
      boolean exclusive=true;
      DefaultConsumer consumer=new DefaultConsumer(channel){
        @Override public void handleDelivery(        String consumerTag,        Envelope envelope,        AMQP.BasicProperties properties,        byte[] body) throws IOException {
          MessageProperties messageProperties=messagePropertiesConverter.toMessageProperties(properties,envelope,encoding);
          Message reply=new Message(body,messageProperties);
          if (logger.isTraceEnabled()) {
            logger.trace("Message received " + reply);
          }
          try {
            replyHandoff.put(reply);
          }
 catch (          InterruptedException e) {
            Thread.currentThread().interrupt();
          }
        }
      }
;
      channel.basicConsume(replyTo,noAck,consumerTag,noLocal,exclusive,null,consumer);
      doSend(channel,exchange,routingKey,message,null);
      Message reply=(replyTimeout < 0) ? replyHandoff.take() : replyHandoff.poll(replyTimeout,TimeUnit.MILLISECONDS);
      channel.basicCancel(consumerTag);
      return reply;
    }
  }
);
  return replyMessage;
}
 

protected void onInit() throws Exception {
  Assert.notNull(bucket,"Providing a valid S3 Bucket name is mandatory");
  Assert.isTrue(directory != null && directory.exists() && directory.isDirectory(),"Please provide a valid local directory to synchronize the remote files with");
  AmazonS3OperationsImpl s3Operations=new AmazonS3OperationsImpl(credentials);
  s3Operations.setTemporaryFileSuffix(temporarySuffix);
  s3Operations.setThreadPoolExecutor(threadPoolExecutor);
  s3Operations.afterPropertiesSet();
  InboundLocalFileOperationsImpl fileOperations=new InboundLocalFileOperationsImpl();
  fileOperations.setTemporaryFileSuffix(temporarySuffix);
  fileOperations.addEventListener(this);
  InboundFileSynchronizationImpl synchronizationImpl=new InboundFileSynchronizationImpl(s3Operations,fileOperations);
  synchronizationImpl.setSynchronizingBatchSize(maxObjectsPerBatch);
  synchronizationImpl.setFileWildcard(fileWildcard);
  synchronizationImpl.setFileNamePattern(fileNameRegex);
  synchronizationImpl.afterPropertiesSet();
  this.synchronizer=synchronizationImpl;
  filesQueue=new ArrayBlockingQueue<File>(queueSize > 0 && queueSize < MAX_QUEUE_CAPACITY ? queueSize : MAX_QUEUE_CAPACITY);
}
 

@Override public void configure(PluginContext context) throws ReplicatorException, InterruptedException {
  this.serviceName=context.getServiceName();
  if (this.handlers == null) {
    throw new ReplicatorException("Handler can not be empty.");
  }
  try {
    this.syncHandlers=new HashMap<String,DBSyncHandler>();
    for (    String handler : this.handlers.split(",")) {
      DBSyncHandler h=(DBSyncHandler)Class.forName(handler).newInstance();
      this.syncHandlers.put(h.interest(),h);
    }
  }
 catch (  Exception e) {
    throw new ReplicatorException("Init handler failed.",e);
  }
  try {
    this.tedisManager=TedisManagerFactory.create(appName,version);
  }
 catch (  Exception e) {
    throw new ReplicatorException("Redis init failed.",e);
  }
  for (int i=0; i < threadSize; i++) {
    processingQueues.add(new ArrayBlockingQueue<DataEvent>(processingQueueLimit));
    Thread thread=new ProcessingThread(i);
    threadPool.add(thread);
    thread.start();
  }
  waitingThread=new WaitingThread();
  waitingThread.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();
}
 

protected void write(SelectionKey key){
  SocketChannel socketChannel=(SocketChannel)key.channel();
  for (Iterator<Entry<Integer,ArrayBlockingQueue<ByteBuffer>>> it=scheduledWrites.entrySet().iterator(); it.hasNext(); ) {
    try {
      Entry<Integer,ArrayBlockingQueue<ByteBuffer>> entry=it.next();
      if (entry.getKey().equals((int)key.attachment())) {
        ArrayBlockingQueue<ByteBuffer> writeOps=entry.getValue();
        for (Iterator<ByteBuffer> itr=writeOps.iterator(); itr.hasNext(); ) {
          ByteBuffer bb=itr.next();
          socketChannel.write(bb);
          if (bb.remaining() > 0) {
            Logger.getLogger(getClass()).debug("Write interrupt on " + (int)key.attachment());
            break;
          }
          itr.remove();
        }
      }
    }
 catch (    IOException ex) {
      Logger.getLogger(getClass()).error("Failed to write data",ex);
    }
  }
  key.interestOps(SelectionKey.OP_READ);
}
 

/** 
 * @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 QuorumCnxManager(QuorumPeer self){
  this.recvQueue=new ArrayBlockingQueue<Message>(CAPACITY);
  this.queueSendMap=new ConcurrentHashMap<Long,ArrayBlockingQueue<ByteBuffer>>();
  this.senderWorkerMap=new ConcurrentHashMap<Long,SendWorker>();
  this.lastMessageSent=new ConcurrentHashMap<Long,ByteBuffer>();
  String cnxToValue=System.getProperty("zookeeper.cnxTimeout");
  if (cnxToValue != null) {
    this.cnxTO=new Integer(cnxToValue);
  }
  this.self=self;
  listener=new Listener();
}
 

public QuorumCnxManager(QuorumPeer self){
  this.recvQueue=new ArrayBlockingQueue<Message>(RECV_CAPACITY);
  this.queueSendMap=new ConcurrentHashMap<Long,ArrayBlockingQueue<ByteBuffer>>();
  this.senderWorkerMap=new ConcurrentHashMap<Long,SendWorker>();
  this.lastMessageSent=new ConcurrentHashMap<Long,ByteBuffer>();
  String cnxToValue=System.getProperty("zookeeper.cnxTimeout");
  if (cnxToValue != null) {
    this.cnxTO=new Integer(cnxToValue);
  }
  this.self=self;
  listener=new Listener();
  listener.setName("QuorumPeerListener");
}