@Test public void enqueuingMaxSizeTriggersBatching() throws Exception { doReturn(1).when(executor).getActiveCount(); doReturn(1).when(executor).getPoolSize(); SingleRollupWriteContext[] srwcs = new SingleRollupWriteContext[100]; int i; for (i = 0; i < 100; i++) { srwcs[i] = mock(SingleRollupWriteContext.class); Points<SimpleNumber> points = new Points<SimpleNumber>(); Rollup rollup = Rollup.BasicFromRaw.compute(points); doReturn(rollup).when(srwcs[i]).getRollup(); } for (i = 0; i < 100; i++) { rbw.enqueueRollupForWrite(srwcs[i]); } verify(ctx, times(100)).incrementWriteCounter(); verifyNoMoreInteractions(ctx); verify(executor, times(96)).getActiveCount(); verify(executor, times(96)).getPoolSize(); verify(executor).execute(Matchers.<Runnable>any()); verifyNoMoreInteractions(executor); }
public void enqueueRollupForWrite(SingleRollupWriteContext rollupWriteContext) { rollupQueue.add(rollupWriteContext); context.incrementWriteCounter(); if (rollupQueue.size() >= ROLLUP_BATCH_MIN_SIZE) { if (executor.getActiveCount() < executor.getPoolSize() || rollupQueue.size() >= ROLLUP_BATCH_MAX_SIZE) { drainBatch(); } } }
RollupBatchWriter { public void enqueueRollupForWrite(SingleRollupWriteContext rollupWriteContext) { rollupQueue.add(rollupWriteContext); context.incrementWriteCounter(); if (rollupQueue.size() >= ROLLUP_BATCH_MIN_SIZE) { if (executor.getActiveCount() < executor.getPoolSize() || rollupQueue.size() >= ROLLUP_BATCH_MAX_SIZE) { drainBatch(); } } } }
RollupBatchWriter { public void enqueueRollupForWrite(SingleRollupWriteContext rollupWriteContext) { rollupQueue.add(rollupWriteContext); context.incrementWriteCounter(); if (rollupQueue.size() >= ROLLUP_BATCH_MIN_SIZE) { if (executor.getActiveCount() < executor.getPoolSize() || rollupQueue.size() >= ROLLUP_BATCH_MAX_SIZE) { drainBatch(); } } } RollupBatchWriter(ThreadPoolExecutor executor, RollupExecutionContext context); }
RollupBatchWriter { public void enqueueRollupForWrite(SingleRollupWriteContext rollupWriteContext) { rollupQueue.add(rollupWriteContext); context.incrementWriteCounter(); if (rollupQueue.size() >= ROLLUP_BATCH_MIN_SIZE) { if (executor.getActiveCount() < executor.getPoolSize() || rollupQueue.size() >= ROLLUP_BATCH_MAX_SIZE) { drainBatch(); } } } RollupBatchWriter(ThreadPoolExecutor executor, RollupExecutionContext context); void enqueueRollupForWrite(SingleRollupWriteContext rollupWriteContext); synchronized void drainBatch(); }
RollupBatchWriter { public void enqueueRollupForWrite(SingleRollupWriteContext rollupWriteContext) { rollupQueue.add(rollupWriteContext); context.incrementWriteCounter(); if (rollupQueue.size() >= ROLLUP_BATCH_MIN_SIZE) { if (executor.getActiveCount() < executor.getPoolSize() || rollupQueue.size() >= ROLLUP_BATCH_MAX_SIZE) { drainBatch(); } } } RollupBatchWriter(ThreadPoolExecutor executor, RollupExecutionContext context); void enqueueRollupForWrite(SingleRollupWriteContext rollupWriteContext); synchronized void drainBatch();  }