Use batch-draining queues in SingleThreadExecutor

[merlimat]

Motivation

SingleThreadExecutor backs every OrderedExecutor thread — including the bookie read/write thread pools — so its queue is on the task-dispatch hot path. Today it uses:

• GrowableMpScArrayConsumerBlockingQueue (unbounded, the default), draining element-by-element via drainTo(ArrayList)
• JDK ArrayBlockingQueue (bounded, when maxTasksInQueue is set)

Meanwhile BatchedArrayBlockingQueue — purpose-built in the same package for the many-producers/one-consumer shape, with System.arraycopy batch draining — was unused by the executor. A JMH comparison (8 producers, consumers running the executor's exact drain loops) showed the growable queue is actually the slowest option, and the batched queue the fastest:

Queue	put throughput
GrowableMpScArrayConsumerBlockingQueue (current unbounded)	32.7 Mops/s
JDK ArrayBlockingQueue (current bounded)	43.3 Mops/s
GrowableBatchedArrayBlockingQueue (new, unbounded)	48.5 Mops/s (+48%)
BatchedArrayBlockingQueue (bounded)	51.4 Mops/s (+19%)

Changes

• Add GrowableBatchedArrayBlockingQueue: the unbounded companion of BatchedArrayBlockingQueue. Same single-lock design and two-span System.arraycopy batch operations (takeAll / pollAll / putAll), but with no notFull condition — producers never block; the backing array doubles when full (and never shrinks), like the queue it replaces.
• SingleThreadExecutor now uses BatchedArrayBlockingQueue for the bounded case and the new growable queue for the unbounded case, and the run loop drains with takeAll into a reusable Runnable[] (capped at 1024) instead of per-element drainTo(ArrayList).
• Remove GrowableMpScArrayConsumerBlockingQueue (and its test): the executor was its only production user, so it is fully superseded by the new queue. The stray usage in BatchedArrayBlockingQueueTest.blockingTake now tests BatchedArrayBlockingQueue itself, as intended.
• New unit test covering the growth paths (including growth and batch-insert with wrapped indexes) and the batched operations.
• New JMH benchmark (SingleThreadExecutorQueueBenchmark) comparing the queue options with executor-style consumers.

Verified: new queue test (9/9), BatchedArrayBlockingQueueTest, TestSingleThreadExecutor (10/10), TestOrderedExecutor, TestOrderedExecutorDecorators, plus BookieJournalTest as an integration smoke through the bookie path (16/16). Checkstyle clean.

[hangc0276]

Issues in GrowableBatchedArrayBlockingQueue

1. drainTo() should validate maxElements

Potential correctness issue: drainTo(Collection<? super T> c, int maxElements) should validate that maxElements is non-negative. Currently, if a negative value is passed, toDrain becomes negative (Math.min(size, maxElements)), and size -= toDrain actually increases the queue size, leaving the internal state inconsistent. It would be safer to reject negative values up front by throwing an IllegalArgumentException, which is also consistent with the BlockingQueue#drainTo contract.

2. drainTo() is not exception-safe

Potential correctness issue: drainTo() is not exception-safe if c.add(item) throws. At that point the queue slot has already been cleared, but consumerIdx and size have not yet been updated, leaving the queue in an inconsistent state. Consider removing the element from the queue only after it has been successfully added to the destination collection, or otherwise ensuring the queue remains consistent if an exception occurs.

3. Prevent draining into the same queue

drainTo() should probably reject c == this. Draining a queue into itself is generally unsupported by the BlockingQueue contract and may lead to unexpected behavior or corrupt the queue state. It would be better to fail fast with an IllegalArgumentException.

4. Validate the initial capacity

The constructor should validate that initialCapacity > 0. If initialCapacity is 0, grow() will never make progress because newCapacity starts at 0 and newCapacity *= 2 remains 0, resulting in an infinite loop on the first insertion.

5. Reject null elements

Should this queue reject null elements? Most Queue/BlockingQueue implementations disallow null because poll() uses null to indicate that the queue is empty. Allowing null makes it impossible for callers to distinguish between an empty queue and a stored null value.

6. Preserve the remaining timeout in pollAll()

internalTakeAll() restarts the full timeout after every wake-up by repeatedly calling await(timeout, unit). In the presence of spurious wake-ups (or if the implementation is ever used with multiple consumers), the total wait time may exceed the requested timeout. It would be more robust to track the remaining timeout using awaitNanos(), similar to the implementation of poll().

7. Validate the arguments to putAll()

putAll() should validate its input arguments (array, offset, and len) before performing any copies. This provides clearer error reporting and avoids partially modifying internal state if invalid arguments are passed.

8. Guard against integer overflow during growth

grow() does not appear to guard against integer overflow when repeatedly doubling the capacity. Similarly, size + len may overflow for very large queues. It would be safer to detect overflow explicitly and fail with a clear exception rather than relying on array allocation failures or overflow behavior.

9. Handle zero-length destination arrays

It may be worth handling the case where array.length == 0 in takeAll()/pollAll(). Currently, these methods may still block waiting for data even though they cannot transfer any elements. Returning immediately with 0 would avoid unnecessary waiting.

10. Consider using signal() instead of signalAll()

The class-level comment states that this queue only supports a single consumer thread. If that assumption always holds, notEmpty.signal() should be sufficient and would avoid waking unnecessary waiters. If multiple consumers are actually supported, it may be worth updating the class documentation to reflect the intended concurrency model.

[void-ptr974]

LGTM, thanks!