@Test public void testLargeQueueDoesNotConsumeTooManyThreads() throws Exception { final int NUM_TIMES = 3000; final CountDownLatch latch = new CountDownLatch(1); BasicTask<Void> blockingTask = new BasicTask<Void>(newLatchAwaiter(latch)); em.submit(MutableMap.of("tag", "category1"), blockingTask); final AtomicInteger counter = new AtomicInteger(0); for (int i = 0; i < NUM_TIMES; i++) { BasicTask<Void> t = new BasicTask<Void>(new Runnable() { public void run() { counter.incrementAndGet(); }}); em.submit(MutableMap.of("tag", "category1"), t); if (i % 500 == 0) log.info("Submitted "+i+" jobs..."); } Thread.sleep(100); latch.countDown(); Asserts.succeedsEventually(new Runnable() { @Override public void run() { assertEquals(counter.get(), NUM_TIMES); }}); }
@Override public synchronized <T> Future<T> submit(Callable<T> c) { if (running.compareAndSet(false, true)) { return executeNow(c); } else { WrappingFuture<T> f = new WrappingFuture<T>(); order.add(new QueuedSubmission<T>(c, f)); queueSize++; if (queueSize>0 && (queueSize == 50 || (queueSize<=500 && (queueSize%100)==0) || (queueSize%1000)==0) && queueSize!=lastSizeWarn) { LOG.warn("{} is backing up, {} tasks queued", this, queueSize); if (LOG.isDebugEnabled()) { LOG.debug("Task queue backing up detail, queue "+this+"; task context is "+Tasks.current()+"; latest task is "+c+"; first task is "+order.peek()); } lastSizeWarn = queueSize; } return f; } }
SingleThreadedScheduler implements TaskScheduler, CanSetName { @Override public synchronized <T> Future<T> submit(Callable<T> c) { if (running.compareAndSet(false, true)) { return executeNow(c); } else { WrappingFuture<T> f = new WrappingFuture<T>(); order.add(new QueuedSubmission<T>(c, f)); queueSize++; if (queueSize>0 && (queueSize == 50 || (queueSize<=500 && (queueSize%100)==0) || (queueSize%1000)==0) && queueSize!=lastSizeWarn) { LOG.warn("{} is backing up, {} tasks queued", this, queueSize); if (LOG.isDebugEnabled()) { LOG.debug("Task queue backing up detail, queue "+this+"; task context is "+Tasks.current()+"; latest task is "+c+"; first task is "+order.peek()); } lastSizeWarn = queueSize; } return f; } } }
SingleThreadedScheduler implements TaskScheduler, CanSetName { @Override public synchronized <T> Future<T> submit(Callable<T> c) { if (running.compareAndSet(false, true)) { return executeNow(c); } else { WrappingFuture<T> f = new WrappingFuture<T>(); order.add(new QueuedSubmission<T>(c, f)); queueSize++; if (queueSize>0 && (queueSize == 50 || (queueSize<=500 && (queueSize%100)==0) || (queueSize%1000)==0) && queueSize!=lastSizeWarn) { LOG.warn("{} is backing up, {} tasks queued", this, queueSize); if (LOG.isDebugEnabled()) { LOG.debug("Task queue backing up detail, queue "+this+"; task context is "+Tasks.current()+"; latest task is "+c+"; first task is "+order.peek()); } lastSizeWarn = queueSize; } return f; } }  }
SingleThreadedScheduler implements TaskScheduler, CanSetName { @Override public synchronized <T> Future<T> submit(Callable<T> c) { if (running.compareAndSet(false, true)) { return executeNow(c); } else { WrappingFuture<T> f = new WrappingFuture<T>(); order.add(new QueuedSubmission<T>(c, f)); queueSize++; if (queueSize>0 && (queueSize == 50 || (queueSize<=500 && (queueSize%100)==0) || (queueSize%1000)==0) && queueSize!=lastSizeWarn) { LOG.warn("{} is backing up, {} tasks queued", this, queueSize); if (LOG.isDebugEnabled()) { LOG.debug("Task queue backing up detail, queue "+this+"; task context is "+Tasks.current()+"; latest task is "+c+"; first task is "+order.peek()); } lastSizeWarn = queueSize; } return f; } }  @Override void setName(String name); @Override String toString(); @Override void injectExecutor(ExecutorService executor); @Override synchronized Future<T> submit(Callable<T> c); }
SingleThreadedScheduler implements TaskScheduler, CanSetName { @Override public synchronized <T> Future<T> submit(Callable<T> c) { if (running.compareAndSet(false, true)) { return executeNow(c); } else { WrappingFuture<T> f = new WrappingFuture<T>(); order.add(new QueuedSubmission<T>(c, f)); queueSize++; if (queueSize>0 && (queueSize == 50 || (queueSize<=500 && (queueSize%100)==0) || (queueSize%1000)==0) && queueSize!=lastSizeWarn) { LOG.warn("{} is backing up, {} tasks queued", this, queueSize); if (LOG.isDebugEnabled()) { LOG.debug("Task queue backing up detail, queue "+this+"; task context is "+Tasks.current()+"; latest task is "+c+"; first task is "+order.peek()); } lastSizeWarn = queueSize; } return f; } }  @Override void setName(String name); @Override String toString(); @Override void injectExecutor(ExecutorService executor); @Override synchronized Future<T> submit(Callable<T> c);  }