Added BatchedArrayBlockingQueue #3838
Conversation
Codecov Report
@@ Coverage Diff @@
## master #3838 +/- ##
============================================
+ Coverage 68.21% 68.32% +0.10%
+ Complexity 6761 6757 -4
============================================
Files 473 473
Lines 40950 40950
Branches 5240 5240
============================================
+ Hits 27935 27980 +45
+ Misses 10762 10715 -47
- Partials 2253 2255 +2
Flags with carried forward coverage won't be shown. Click here to find out more.
📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more |
| } | ||
|
|
||
| public int putAll(List<T> c) throws InterruptedException { | ||
| lock.lockInterruptibly(); |
There was a problem hiding this comment.
Do we need to check whether the items in the list is null?
| } | ||
|
|
||
| this.consumerIdx = consumerIdx; | ||
| if (size == capacity) { |
There was a problem hiding this comment.
If the maxElements is 0, we shouldn't trigger notFull.signalAll();. Shall we check the maxElements param?
| } | ||
|
|
||
| @Override | ||
| public int remainingCapacity() { |
There was a problem hiding this comment.
The method size() is locked, and the method remainingCapacity should also be locked.
| this.producerIdx = producerIdx; | ||
|
|
||
| if (size == 0) { | ||
| notEmpty.signalAll(); |
There was a problem hiding this comment.
No needn't trigger notEmpty.signalAll(); if the target collection c is empty.
|
|
||
| // First span | ||
| int firstSpan = Math.min(toInsert, capacity - producerIdx); | ||
| System.arraycopy(a, offset, data, producerIdx, firstSpan); |
There was a problem hiding this comment.
It may throw ArrayIndexOutOfBoundsException when the param is illegal.
Integer[] integers = new Integer[] {1, 2, 3, 4, 5, 6, 7, 8, 9,10,11,12,13,14,15,16,17,18,19,20};
queue.putAll(integers, 10, 11);
| } | ||
|
|
||
| @Override | ||
| public void clear() { |
There was a problem hiding this comment.
And the clear() may not reach the expectation, it may can't clear the queue.
new Thread(()-> {
Integer[] integers = new Integer[] {1, 2, 3, 4, 5, 6, 7, 8, 9,10,11,12,13,14,15,16,17,18,19,20};
try {
queue.putAll(integers, 0, 20);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}).start();
queue.clear();
//After clear, the queue is not empty.
System.out.println(queue.size());
Motivation
One of the main contention points in bookies is the Journal queue. There are multiple threads posting entries to be added on the journal on the queue.
In order to reduce contention and increase overall throughput (when there are many small entries), we can adopt a strategy of "bulk" inserting to the queue.
BatchedArrayBlockingQueueis a fixed size blocking queue, very similar to JavaArrayBlockingQueuewith 2 additional methods:putAll(T[] array, int offset, int len)--> Add all the items from the array in one shot, block if fulltakeAll(T[] dest)--> Take all the available items in the queue that fit into the destination array, block if emptyWe use arrays here instead of List/Collection because it enables to do
System.arrayCopy()which is way faster than a for loop and assignments.This PR only introduces the queue, subsequent work will make use of it for the journal.
Benchmarks
When inserting items in bulk, this queue is 10x to 20x faster than
ArrayBlockingQueue