This is an automated email from the ASF dual-hosted git repository.
merlimat pushed a commit to branch branch-4.18
in repository https://gitbox.apache.org/repos/asf/bookkeeper.git
The following commit(s) were added to refs/heads/branch-4.18 by this push:
new 06e9409b0d Use batch-draining queues in SingleThreadExecutor (#4836)
06e9409b0d is described below
commit 06e9409b0d3b7eca33c6a42c1dfd4a536c24e1e3
Author: Matteo Merli <[email protected]>
AuthorDate: Thu Jul 16 09:25:18 2026 -0700
Use batch-draining queues in SingleThreadExecutor (#4836)
* Use batch-draining queues in SingleThreadExecutor
* Remove GrowableMpScArrayConsumerBlockingQueue
* Address review comments in GrowableBatchedArrayBlockingQueue
(cherry picked from commit 68cc8dcbd1e8a7e95c68e70d183e178ad6d84ede)
---
.../GrowableBatchedArrayBlockingQueue.java | 401 +++++++++++++++++++++
.../GrowableMpScArrayConsumerBlockingQueue.java | 331 -----------------
.../common/util/SingleThreadExecutor.java | 41 ++-
.../collections/BatchedArrayBlockingQueueTest.java | 2 +-
.../GrowableArrayBlockingQueueTest.java | 273 --------------
.../GrowableBatchedArrayBlockingQueueTest.java | 380 +++++++++++++++++++
.../common/SingleThreadExecutorQueueBenchmark.java | 139 +++++++
7 files changed, 942 insertions(+), 625 deletions(-)
diff --git
a/bookkeeper-common/src/main/java/org/apache/bookkeeper/common/collections/GrowableBatchedArrayBlockingQueue.java
b/bookkeeper-common/src/main/java/org/apache/bookkeeper/common/collections/GrowableBatchedArrayBlockingQueue.java
new file mode 100644
index 0000000000..2135096002
--- /dev/null
+++
b/bookkeeper-common/src/main/java/org/apache/bookkeeper/common/collections/GrowableBatchedArrayBlockingQueue.java
@@ -0,0 +1,401 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.bookkeeper.common.collections;
+
+import java.util.AbstractQueue;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.Objects;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
+import org.apache.bookkeeper.common.util.MathUtils;
+
+/**
+ * This implements a {@link BlockingQueue} backed by an array with no fixed
capacity.
+ *
+ * <p>It is the unbounded companion of {@link BatchedArrayBlockingQueue}: when
the backing array
+ * is full it is doubled in size (and never shrunk), so producers never block.
+ *
+ * <p>This queue only allows 1 consumer thread to dequeue items and multiple
producer threads.
+ */
+public class GrowableBatchedArrayBlockingQueue<T>
+ extends AbstractQueue<T>
+ implements BlockingQueue<T>, BatchedBlockingQueue<T> {
+
+ private final ReentrantLock lock = new ReentrantLock();
+
+ private final Condition notEmpty = lock.newCondition();
+
+ private T[] data;
+
+ private int size;
+
+ private int consumerIdx;
+ private int producerIdx;
+
+ public GrowableBatchedArrayBlockingQueue() {
+ this(64);
+ }
+
+ @SuppressWarnings("unchecked")
+ public GrowableBatchedArrayBlockingQueue(int initialCapacity) {
+ int capacity = MathUtils.findNextPositivePowerOfTwo(initialCapacity);
+ data = (T[]) new Object[capacity];
+ }
+
+ private T dequeueOne() {
+ T item = data[consumerIdx];
+ data[consumerIdx] = null;
+ if (++consumerIdx == data.length) {
+ consumerIdx = 0;
+ }
+
+ --size;
+ return item;
+ }
+
+ private void enqueueOne(T item) {
+ if (size == data.length) {
+ grow(size + 1);
+ }
+
+ data[producerIdx] = item;
+ if (++producerIdx == data.length) {
+ producerIdx = 0;
+ }
+
+ if (size++ == 0) {
+ // There is a single consumer thread, so no need to use signalAll()
+ notEmpty.signal();
+ }
+ }
+
+ // must be called while holding the lock
+ @SuppressWarnings("unchecked")
+ private void grow(int minCapacity) {
+ if (minCapacity < 0) {
+ // The requested capacity overflowed the int range
+ throw new IllegalStateException("Queue capacity would exceed the
maximum array size");
+ }
+
+ int newCapacity = data.length;
+ while (newCapacity < minCapacity) {
+ newCapacity *= 2;
+ if (newCapacity <= 0) {
+ // Doubling overflowed: fall back to the exact requested
capacity
+ newCapacity = minCapacity;
+ }
+ }
+
+ T[] newData = (T[]) new Object[newCapacity];
+
+ int firstSpan = Math.min(size, data.length - consumerIdx);
+ System.arraycopy(data, consumerIdx, newData, 0, firstSpan);
+
+ int secondSpan = size - firstSpan;
+ if (secondSpan > 0) {
+ System.arraycopy(data, 0, newData, firstSpan, secondSpan);
+ }
+
+ data = newData;
+ consumerIdx = 0;
+ producerIdx = size;
+ }
+
+ @Override
+ public T poll() {
+ lock.lock();
+
+ try {
+ if (size == 0) {
+ return null;
+ }
+
+ return dequeueOne();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public T peek() {
+ lock.lock();
+
+ try {
+ if (size == 0) {
+ return null;
+ }
+
+ return data[consumerIdx];
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public boolean offer(T e) {
+ Objects.requireNonNull(e);
+
+ lock.lock();
+
+ try {
+ enqueueOne(e);
+
+ return true;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public void put(T e) {
+ offer(e);
+ }
+
+ @Override
+ public boolean offer(T e, long timeout, TimeUnit unit) {
+ // Queue is unbounded and it will never reject new items
+ return offer(e);
+ }
+
+ @Override
+ public void putAll(T[] a, int offset, int len) {
+ Objects.requireNonNull(a);
+ Objects.checkFromIndexSize(offset, len, a.length);
+
+ lock.lock();
+
+ try {
+ if (len > data.length - size) {
+ grow(size + len);
+ }
+
+ int capacity = data.length;
+ int producerIdx = this.producerIdx;
+
+ // First span
+ int firstSpan = Math.min(len, capacity - producerIdx);
+ System.arraycopy(a, offset, data, producerIdx, firstSpan);
+ producerIdx += firstSpan;
+
+ int secondSpan = len - firstSpan;
+ if (secondSpan > 0) {
+ System.arraycopy(a, offset + firstSpan, data, 0, secondSpan);
+ producerIdx = secondSpan;
+ }
+
+ if (producerIdx == capacity) {
+ producerIdx = 0;
+ }
+
+ this.producerIdx = producerIdx;
+
+ if (size == 0) {
+ // There is a single consumer thread, so no need to use
signalAll()
+ notEmpty.signal();
+ }
+
+ size += len;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public T take() throws InterruptedException {
+ lock.lockInterruptibly();
+
+ try {
+ while (size == 0) {
+ notEmpty.await();
+ }
+
+ return dequeueOne();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public T poll(long timeout, TimeUnit unit) throws InterruptedException {
+ long remainingTimeNanos = unit.toNanos(timeout);
+
+ lock.lockInterruptibly();
+ try {
+ while (size == 0) {
+ if (remainingTimeNanos <= 0L) {
+ return null;
+ }
+
+ remainingTimeNanos = notEmpty.awaitNanos(remainingTimeNanos);
+ }
+
+ return dequeueOne();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public int remainingCapacity() {
+ return Integer.MAX_VALUE;
+ }
+
+ @Override
+ public int drainTo(Collection<? super T> c) {
+ return drainTo(c, Integer.MAX_VALUE);
+ }
+
+ @Override
+ public int drainTo(Collection<? super T> c, int maxElements) {
+ Objects.requireNonNull(c);
+ if (c == this) {
+ throw new IllegalArgumentException("Cannot drain a queue into
itself");
+ }
+ if (maxElements <= 0) {
+ return 0;
+ }
+
+ lock.lock();
+ try {
+ int toDrain = Math.min(size, maxElements);
+
+ int capacity = data.length;
+ int consumerIdx = this.consumerIdx;
+ int drained = 0;
+
+ try {
+ while (drained < toDrain) {
+ T item = data[consumerIdx];
+ c.add(item);
+
+ // Only clear the slot once the item was accepted by the
target collection
+ data[consumerIdx] = null;
+ if (++consumerIdx == capacity) {
+ consumerIdx = 0;
+ }
+ ++drained;
+ }
+ } finally {
+ // Even if c.add() threw, commit the items that were actually
transferred
+ this.consumerIdx = consumerIdx;
+ size -= drained;
+ }
+
+ return drained;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public int takeAll(T[] array) throws InterruptedException {
+ return internalTakeAll(array, true, 0, TimeUnit.SECONDS);
+ }
+
+ @Override
+ public int pollAll(T[] array, long timeout, TimeUnit unit) throws
InterruptedException {
+ return internalTakeAll(array, false, timeout, unit);
+ }
+
+ private int internalTakeAll(T[] array, boolean waitForever, long timeout,
TimeUnit unit)
+ throws InterruptedException {
+ if (array.length == 0) {
+ return 0;
+ }
+
+ long remainingTimeNanos = unit.toNanos(timeout);
+
+ lock.lockInterruptibly();
+ try {
+ while (size == 0) {
+ if (waitForever) {
+ notEmpty.await();
+ } else {
+ if (remainingTimeNanos <= 0L) {
+ return 0;
+ }
+
+ remainingTimeNanos =
notEmpty.awaitNanos(remainingTimeNanos);
+ }
+ }
+
+ int toDrain = Math.min(size, array.length);
+
+ int capacity = data.length;
+ int consumerIdx = this.consumerIdx;
+
+ // First span
+ int firstSpan = Math.min(toDrain, capacity - consumerIdx);
+ System.arraycopy(data, consumerIdx, array, 0, firstSpan);
+ Arrays.fill(data, consumerIdx, consumerIdx + firstSpan, null);
+ consumerIdx += firstSpan;
+
+ int secondSpan = toDrain - firstSpan;
+ if (secondSpan > 0) {
+ System.arraycopy(data, 0, array, firstSpan, secondSpan);
+ Arrays.fill(data, 0, secondSpan, null);
+ consumerIdx = secondSpan;
+ }
+
+ if (consumerIdx == capacity) {
+ consumerIdx = 0;
+ }
+ this.consumerIdx = consumerIdx;
+
+ size -= toDrain;
+ return toDrain;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public void clear() {
+ lock.lock();
+ try {
+ while (size > 0) {
+ dequeueOne();
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public int size() {
+ lock.lock();
+
+ try {
+ return size;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public Iterator<T> iterator() {
+ throw new UnsupportedOperationException();
+ }
+}
diff --git
a/bookkeeper-common/src/main/java/org/apache/bookkeeper/common/collections/GrowableMpScArrayConsumerBlockingQueue.java
b/bookkeeper-common/src/main/java/org/apache/bookkeeper/common/collections/GrowableMpScArrayConsumerBlockingQueue.java
deleted file mode 100644
index 1e614f1820..0000000000
---
a/bookkeeper-common/src/main/java/org/apache/bookkeeper/common/collections/GrowableMpScArrayConsumerBlockingQueue.java
+++ /dev/null
@@ -1,331 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-package org.apache.bookkeeper.common.collections;
-
-import java.util.AbstractQueue;
-import java.util.Collection;
-import java.util.Iterator;
-import java.util.NoSuchElementException;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.locks.LockSupport;
-import java.util.concurrent.locks.StampedLock;
-import org.apache.bookkeeper.common.util.MathUtils;
-
-
-/**
- * This implements a {@link BlockingQueue} backed by an array with no fixed
capacity.
- *
- * <p>When the capacity is reached, data will be moved to a bigger array.
- *
- * <p>This queue only allows 1 consumer thread to dequeue items and multiple
producer threads.
- */
-public class GrowableMpScArrayConsumerBlockingQueue<T> extends
AbstractQueue<T> implements BlockingQueue<T> {
-
- private final StampedLock headLock = new StampedLock();
- private final PaddedInt headIndex = new PaddedInt();
- private final PaddedInt tailIndex = new PaddedInt();
- private final StampedLock tailLock = new StampedLock();
-
- private T[] data;
- private final AtomicInteger size = new AtomicInteger(0);
-
- private volatile Thread waitingConsumer;
-
- public GrowableMpScArrayConsumerBlockingQueue() {
- this(64);
- }
-
- @SuppressWarnings("unchecked")
- public GrowableMpScArrayConsumerBlockingQueue(int initialCapacity) {
- int capacity = MathUtils.findNextPositivePowerOfTwo(initialCapacity);
- data = (T[]) new Object[capacity];
- }
-
- @Override
- public T remove() {
- T item = poll();
- if (item == null) {
- throw new NoSuchElementException();
- }
-
- return item;
- }
-
- @Override
- public T poll() {
- if (size.get() > 0) {
- // Since this is a single-consumer queue, we don't expect multiple
threads calling poll(), though we need
- // to protect against array expansions
- long stamp = headLock.readLock();
-
- try {
- T item = data[headIndex.value];
- data[headIndex.value] = null;
- headIndex.value = (headIndex.value + 1) & (data.length - 1);
- size.decrementAndGet();
- return item;
- } finally {
- headLock.unlockRead(stamp);
- }
- } else {
- return null;
- }
- }
-
- @Override
- public T element() {
- T item = peek();
- if (item == null) {
- throw new NoSuchElementException();
- }
-
- return item;
- }
-
- @Override
- public T peek() {
- if (size.get() > 0) {
- long stamp = headLock.readLock();
-
- try {
- return data[headIndex.value];
- } finally {
- headLock.unlockRead(stamp);
- }
- } else {
- return null;
- }
- }
-
- @Override
- public boolean offer(T e) {
- // Queue is unbounded and it will never reject new items
- put(e);
- return true;
- }
-
- @Override
- public void put(T e) {
- long stamp = tailLock.writeLock();
-
- try {
- int oldSize = size.get();
- if (oldSize == data.length) {
- expandArray();
- }
-
- data[tailIndex.value] = e;
- tailIndex.value = (tailIndex.value + 1) & (data.length - 1);
-
- if (size.getAndIncrement() == 0 && waitingConsumer != null) {
- Thread waitingConsumer = this.waitingConsumer;
- this.waitingConsumer = null;
- LockSupport.unpark(waitingConsumer);
- }
- } finally {
- tailLock.unlockWrite(stamp);
- }
- }
-
- @Override
- public boolean add(T e) {
- put(e);
- return true;
- }
-
- @Override
- public boolean offer(T e, long timeout, TimeUnit unit) {
- // Queue is unbounded and it will never reject new items
- put(e);
- return true;
- }
-
- @Override
- public T take() throws InterruptedException {
- while (size() == 0) {
- waitingConsumer = Thread.currentThread();
-
- // Double check that size has not changed after we have registered
ourselves for notification
- if (size() == 0) {
- LockSupport.park();
- if (Thread.interrupted()) {
- throw new InterruptedException();
- }
- }
- }
-
- return poll();
- }
-
- @Override
- public T poll(long timeout, TimeUnit unit) throws InterruptedException {
- long deadline = System.currentTimeMillis() + unit.toMillis(timeout);
-
- while (size.get() == 0) {
- waitingConsumer = Thread.currentThread();
-
- // Double check that size has not changed after we have registered
ourselves for notification
- if (size.get() == 0) {
- LockSupport.parkUntil(deadline);
- if (Thread.interrupted()) {
- throw new InterruptedException();
- }
-
- if (System.currentTimeMillis() >= deadline) {
- return null;
- }
- }
- }
-
- return poll();
- }
-
- @Override
- public int remainingCapacity() {
- return Integer.MAX_VALUE;
- }
-
- @Override
- public int drainTo(Collection<? super T> c) {
- return drainTo(c, Integer.MAX_VALUE);
- }
-
- @Override
- public int drainTo(Collection<? super T> c, int maxElements) {
- long stamp = headLock.readLock();
-
- try {
- int toDrain = Math.min(size.get(), maxElements);
-
- for (int i = 0; i < toDrain; i++) {
- T item = data[headIndex.value];
- data[headIndex.value] = null;
- c.add(item);
-
- headIndex.value = (headIndex.value + 1) & (data.length - 1);
- }
-
- this.size.addAndGet(-toDrain);
- return toDrain;
- } finally {
- headLock.unlockRead(stamp);
- }
- }
-
- @Override
- public void clear() {
- long stamp = headLock.readLock();
-
- try {
- int size = this.size.get();
-
- for (int i = 0; i < size; i++) {
- data[headIndex.value] = null;
- headIndex.value = (headIndex.value + 1) & (data.length - 1);
- }
-
- this.size.addAndGet(-size);
- } finally {
- headLock.unlockRead(stamp);
- }
- }
-
- @Override
- public int size() {
- return size.get();
- }
-
- @Override
- public Iterator<T> iterator() {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public String toString() {
- StringBuilder sb = new StringBuilder();
-
- long tailStamp = tailLock.writeLock();
- long headStamp = headLock.writeLock();
-
- try {
- int headIndex = this.headIndex.value;
- int size = this.size.get();
-
- sb.append('[');
-
- for (int i = 0; i < size; i++) {
- T item = data[headIndex];
- if (i > 0) {
- sb.append(", ");
- }
-
- sb.append(item);
-
- headIndex = (headIndex + 1) & (data.length - 1);
- }
-
- sb.append(']');
- } finally {
- headLock.unlockWrite(headStamp);
- tailLock.unlockWrite(tailStamp);
- }
- return sb.toString();
- }
-
- @SuppressWarnings("unchecked")
- private void expandArray() {
- // We already hold the tailLock
- long headLockStamp = headLock.writeLock();
-
- try {
- int size = this.size.get();
- int newCapacity = data.length * 2;
- T[] newData = (T[]) new Object[newCapacity];
-
-
- int oldHeadIndex = headIndex.value;
- int lenHeadToEnd = Math.min(size, data.length - oldHeadIndex);
-
- System.arraycopy(data, oldHeadIndex, newData, 0, lenHeadToEnd);
- System.arraycopy(data, 0, newData, lenHeadToEnd, size -
lenHeadToEnd);
-
- data = newData;
- headIndex.value = 0;
- tailIndex.value = size;
- } finally {
- headLock.unlockWrite(headLockStamp);
- }
- }
-
- private static final class PaddedInt {
- int value = 0;
-
- // Padding to avoid false sharing
- public volatile int pi1 = 1;
- public volatile long p1 = 1L, p2 = 2L, p3 = 3L, p4 = 4L, p5 = 5L, p6 =
6L;
-
- public long exposeToAvoidOptimization() {
- return pi1 + p1 + p2 + p3 + p4 + p5 + p6;
- }
- }
-}
diff --git
a/bookkeeper-common/src/main/java/org/apache/bookkeeper/common/util/SingleThreadExecutor.java
b/bookkeeper-common/src/main/java/org/apache/bookkeeper/common/util/SingleThreadExecutor.java
index 31cb8aef32..840a4c6dec 100644
---
a/bookkeeper-common/src/main/java/org/apache/bookkeeper/common/util/SingleThreadExecutor.java
+++
b/bookkeeper-common/src/main/java/org/apache/bookkeeper/common/util/SingleThreadExecutor.java
@@ -23,8 +23,6 @@ import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.AbstractExecutorService;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
@@ -34,7 +32,9 @@ import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.LongAdder;
import lombok.CustomLog;
import lombok.SneakyThrows;
-import
org.apache.bookkeeper.common.collections.GrowableMpScArrayConsumerBlockingQueue;
+import org.apache.bookkeeper.common.collections.BatchedArrayBlockingQueue;
+import org.apache.bookkeeper.common.collections.BatchedBlockingQueue;
+import
org.apache.bookkeeper.common.collections.GrowableBatchedArrayBlockingQueue;
import org.apache.bookkeeper.stats.Gauge;
import org.apache.bookkeeper.stats.StatsLogger;
@@ -46,7 +46,10 @@ import org.apache.bookkeeper.stats.StatsLogger;
*/
@CustomLog
public class SingleThreadExecutor extends AbstractExecutorService implements
ExecutorService, Runnable {
- private final BlockingQueue<Runnable> queue;
+
+ private static final int MAX_DRAIN_BATCH_SIZE = 1024;
+
+ private final BatchedBlockingQueue<Runnable> queue;
private final Thread runner;
private final boolean rejectExecution;
@@ -83,9 +86,9 @@ public class SingleThreadExecutor extends
AbstractExecutorService implements Exe
}
if (maxQueueCapacity > 0) {
- this.queue = new ArrayBlockingQueue<>(maxQueueCapacity);
+ this.queue = new BatchedArrayBlockingQueue<>(maxQueueCapacity);
} else {
- this.queue = new GrowableMpScArrayConsumerBlockingQueue<>();
+ this.queue = new GrowableBatchedArrayBlockingQueue<>();
}
this.maxQueueCapacity = maxQueueCapacity;
@@ -102,7 +105,10 @@ public class SingleThreadExecutor extends
AbstractExecutorService implements Exe
public void run() {
try {
boolean isInitialized = false;
- List<Runnable> localTasks = new ArrayList<>();
+ int batchSize = maxQueueCapacity > 0
+ ? Math.min(maxQueueCapacity, MAX_DRAIN_BATCH_SIZE)
+ : MAX_DRAIN_BATCH_SIZE;
+ Runnable[] localTasks = new Runnable[batchSize];
while (state == State.Running) {
if (!isInitialized) {
@@ -110,25 +116,20 @@ public class SingleThreadExecutor extends
AbstractExecutorService implements Exe
isInitialized = true;
}
- int n = queue.drainTo(localTasks);
- if (n > 0) {
- for (int i = 0; i < n; i++) {
- if (!safeRunTask(localTasks.get(i))) {
- return;
- }
- }
- localTasks.clear();
- } else {
- if (!safeRunTask(queue.take())) {
+ int n = queue.takeAll(localTasks);
+ for (int i = 0; i < n; i++) {
+ Runnable task = localTasks[i];
+ localTasks[i] = null;
+ if (!safeRunTask(task)) {
return;
}
}
}
// Clear the queue in orderly shutdown
- int n = queue.drainTo(localTasks);
- for (int i = 0; i < n; i++) {
- safeRunTask(localTasks.get(i));
+ Runnable task;
+ while ((task = queue.poll()) != null) {
+ safeRunTask(task);
}
} catch (InterruptedException ie) {
// Exit loop when interrupted
diff --git
a/bookkeeper-common/src/test/java/org/apache/bookkeeper/common/collections/BatchedArrayBlockingQueueTest.java
b/bookkeeper-common/src/test/java/org/apache/bookkeeper/common/collections/BatchedArrayBlockingQueueTest.java
index 20e2f3723f..a20cc1b470 100644
---
a/bookkeeper-common/src/test/java/org/apache/bookkeeper/common/collections/BatchedArrayBlockingQueueTest.java
+++
b/bookkeeper-common/src/test/java/org/apache/bookkeeper/common/collections/BatchedArrayBlockingQueueTest.java
@@ -94,7 +94,7 @@ public class BatchedArrayBlockingQueueTest {
@Test
public void blockingTake() throws Exception {
- BlockingQueue<Integer> queue = new
GrowableMpScArrayConsumerBlockingQueue<>();
+ BlockingQueue<Integer> queue = new BatchedArrayBlockingQueue<>(100);
CountDownLatch latch = new CountDownLatch(1);
diff --git
a/bookkeeper-common/src/test/java/org/apache/bookkeeper/common/collections/GrowableArrayBlockingQueueTest.java
b/bookkeeper-common/src/test/java/org/apache/bookkeeper/common/collections/GrowableArrayBlockingQueueTest.java
deleted file mode 100644
index 7b20294d58..0000000000
---
a/bookkeeper-common/src/test/java/org/apache/bookkeeper/common/collections/GrowableArrayBlockingQueueTest.java
+++ /dev/null
@@ -1,273 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-package org.apache.bookkeeper.common.collections;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
-import com.google.common.collect.Lists;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.NoSuchElementException;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
-import org.junit.Test;
-
-/**
- * Test the growable array blocking queue.
- */
-public class GrowableArrayBlockingQueueTest {
-
- @Test
- public void simple() throws Exception {
- BlockingQueue<Integer> queue = new
GrowableMpScArrayConsumerBlockingQueue<>(4);
-
- assertEquals(null, queue.poll());
-
- assertEquals(Integer.MAX_VALUE, queue.remainingCapacity());
- assertEquals("[]", queue.toString());
-
- try {
- queue.element();
- fail("Should have thrown exception");
- } catch (NoSuchElementException e) {
- // Expected
- }
-
- try {
- queue.iterator();
- fail("Should have thrown exception");
- } catch (UnsupportedOperationException e) {
- // Expected
- }
-
- // Test index rollover
- for (int i = 0; i < 100; i++) {
- queue.add(i);
-
- assertEquals(i, queue.take().intValue());
- }
-
- queue.offer(1);
- assertEquals("[1]", queue.toString());
- queue.offer(2);
- assertEquals("[1, 2]", queue.toString());
- queue.offer(3);
- assertEquals("[1, 2, 3]", queue.toString());
- queue.offer(4);
- assertEquals("[1, 2, 3, 4]", queue.toString());
-
- assertEquals(4, queue.size());
-
- List<Integer> list = new ArrayList<>();
- queue.drainTo(list, 3);
-
- assertEquals(1, queue.size());
- assertEquals(Lists.newArrayList(1, 2, 3), list);
- assertEquals("[4]", queue.toString());
- assertEquals(4, queue.peek().intValue());
-
- assertEquals(4, queue.element().intValue());
- assertEquals(4, queue.remove().intValue());
- try {
- queue.remove();
- fail("Should have thrown exception");
- } catch (NoSuchElementException e) {
- // Expected
- }
- }
-
- @Test
- public void blockingTake() throws Exception {
- BlockingQueue<Integer> queue = new
GrowableMpScArrayConsumerBlockingQueue<>();
-
- CountDownLatch latch = new CountDownLatch(1);
-
- new Thread(() -> {
- try {
- int expected = 0;
-
- for (int i = 0; i < 100; i++) {
- int n = queue.take();
-
- assertEquals(expected++, n);
- }
-
- latch.countDown();
- } catch (Exception e) {
- e.printStackTrace();
- }
- }).start();
-
- int n = 0;
- for (int i = 0; i < 10; i++) {
- for (int j = 0; j < 10; j++) {
- queue.put(n);
- ++n;
- }
-
- // Wait until all the entries are consumed
- while (!queue.isEmpty()) {
- Thread.sleep(1);
- }
- }
-
- latch.await();
- }
-
- @Test
- public void growArray() throws Exception {
- BlockingQueue<Integer> queue = new
GrowableMpScArrayConsumerBlockingQueue<>(4);
-
- assertEquals(null, queue.poll());
-
- assertTrue(queue.offer(1));
- assertTrue(queue.offer(2));
- assertTrue(queue.offer(3));
- assertTrue(queue.offer(4));
- assertTrue(queue.offer(5));
-
- assertEquals(5, queue.size());
-
- queue.clear();
- assertEquals(0, queue.size());
-
- assertTrue(queue.offer(1, 1, TimeUnit.SECONDS));
- assertTrue(queue.offer(2, 1, TimeUnit.SECONDS));
- assertTrue(queue.offer(3, 1, TimeUnit.SECONDS));
- assertEquals(3, queue.size());
-
- List<Integer> list = new ArrayList<>();
- queue.drainTo(list);
- assertEquals(0, queue.size());
-
- assertEquals(Lists.newArrayList(1, 2, 3), list);
- }
-
- @Test
- public void pollTimeout() throws Exception {
- BlockingQueue<Integer> queue = new
GrowableMpScArrayConsumerBlockingQueue<>(4);
-
- assertEquals(null, queue.poll(1, TimeUnit.MILLISECONDS));
-
- queue.put(1);
- assertEquals(1, queue.poll(1, TimeUnit.MILLISECONDS).intValue());
-
- // 0 timeout should not block
- assertEquals(null, queue.poll(0, TimeUnit.HOURS));
-
- queue.put(2);
- queue.put(3);
- assertEquals(2, queue.poll(1, TimeUnit.HOURS).intValue());
- assertEquals(3, queue.poll(1, TimeUnit.HOURS).intValue());
- }
-
- @Test
- public void pollTimeout2() throws Exception {
- BlockingQueue<Integer> queue = new
GrowableMpScArrayConsumerBlockingQueue<>();
-
- CountDownLatch latch = new CountDownLatch(1);
-
- new Thread(() -> {
- try {
- queue.poll(1, TimeUnit.HOURS);
-
- latch.countDown();
- } catch (Exception e) {
- e.printStackTrace();
- }
- }).start();
-
- // Make sure background thread is waiting on poll
- Thread.sleep(100);
- queue.put(1);
-
- latch.await();
- }
-
-
- static class TestThread extends Thread {
-
- private volatile boolean stop;
- private final BlockingQueue<Integer> readQ;
- private final BlockingQueue<Integer> writeQ;
-
- private final AtomicLong counter = new AtomicLong();
-
- TestThread(BlockingQueue<Integer> readQ, BlockingQueue<Integer>
writeQ) {
- this.readQ = readQ;
- this.writeQ = writeQ;
- }
-
- @Override
- public void run() {
- ArrayList<Integer> localQ = new ArrayList<>();
-
- while (!stop) {
- int items = readQ.drainTo(localQ);
- if (items > 0) {
- for (int i = 0; i < items; i++) {
- writeQ.add(localQ.get(i));
- }
-
- counter.addAndGet(items);
- localQ.clear();
- } else {
- try {
- writeQ.add(readQ.take());
- counter.incrementAndGet();
- } catch (InterruptedException e) {
- return;
- }
- }
- }
- }
- }
-
- public static void main(String[] args) throws Exception {
- int n = 10_000;
- BlockingQueue<Integer> q1 = new
GrowableMpScArrayConsumerBlockingQueue<>();
- BlockingQueue<Integer> q2 = new
GrowableMpScArrayConsumerBlockingQueue<>();
-// BlockingQueue<Integer> q1 = new ArrayBlockingQueue<>(N * 2);
-// BlockingQueue<Integer> q2 = new ArrayBlockingQueue<>(N * 2);
-// BlockingQueue<Integer> q1 = new LinkedBlockingQueue<>();
-// BlockingQueue<Integer> q2 = new LinkedBlockingDeque<>();
-
- TestThread t1 = new TestThread(q1, q2);
- TestThread t2 = new TestThread(q2, q1);
-
- for (int i = 0; i < n; i++) {
- q1.add(i);
- }
-
- t1.start();
- t2.start();
-
- Thread.sleep(10_000);
-
- System.out.println("Throughput " + (t1.counter.get() / 10 / 1e6) + "
Millions items/s");
- t1.stop = true;
- t2.stop = true;
- }
-}
diff --git
a/bookkeeper-common/src/test/java/org/apache/bookkeeper/common/collections/GrowableBatchedArrayBlockingQueueTest.java
b/bookkeeper-common/src/test/java/org/apache/bookkeeper/common/collections/GrowableBatchedArrayBlockingQueueTest.java
new file mode 100644
index 0000000000..c1a6c96d8b
--- /dev/null
+++
b/bookkeeper-common/src/test/java/org/apache/bookkeeper/common/collections/GrowableBatchedArrayBlockingQueueTest.java
@@ -0,0 +1,380 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.bookkeeper.common.collections;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import com.google.common.collect.Lists;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.junit.Test;
+
+/**
+ * Test the growable batched array blocking queue.
+ */
+public class GrowableBatchedArrayBlockingQueueTest {
+
+ @Test
+ public void simple() throws Exception {
+ GrowableBatchedArrayBlockingQueue<Integer> queue = new
GrowableBatchedArrayBlockingQueue<>(4);
+
+ assertNull(queue.poll());
+
+ assertEquals(Integer.MAX_VALUE, queue.remainingCapacity());
+
+ try {
+ queue.element();
+ fail("Should have thrown exception");
+ } catch (NoSuchElementException e) {
+ // Expected
+ }
+
+ try {
+ queue.iterator();
+ fail("Should have thrown exception");
+ } catch (UnsupportedOperationException e) {
+ // Expected
+ }
+
+ // Test index rollover
+ for (int i = 0; i < 100; i++) {
+ queue.add(i);
+
+ assertEquals(i, queue.take().intValue());
+ }
+
+ queue.offer(1);
+ queue.offer(2);
+ queue.offer(3);
+ queue.offer(4);
+
+ assertEquals(4, queue.size());
+
+ List<Integer> list = new ArrayList<>();
+ queue.drainTo(list, 3);
+
+ assertEquals(1, queue.size());
+ assertEquals(Lists.newArrayList(1, 2, 3), list);
+ assertEquals(4, queue.peek().intValue());
+
+ assertEquals(4, queue.element().intValue());
+ assertEquals(4, queue.remove().intValue());
+ try {
+ queue.remove();
+ fail("Should have thrown exception");
+ } catch (NoSuchElementException e) {
+ // Expected
+ }
+ }
+
+ @Test
+ public void blockingTake() throws Exception {
+ GrowableBatchedArrayBlockingQueue<Integer> queue = new
GrowableBatchedArrayBlockingQueue<>();
+
+ CountDownLatch latch = new CountDownLatch(1);
+
+ new Thread(() -> {
+ try {
+ int expected = 0;
+
+ for (int i = 0; i < 100; i++) {
+ int n = queue.take();
+
+ assertEquals(expected++, n);
+ }
+
+ latch.countDown();
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }).start();
+
+ int n = 0;
+ for (int i = 0; i < 10; i++) {
+ for (int j = 0; j < 10; j++) {
+ queue.put(n);
+ ++n;
+ }
+
+ // Wait until all the entries are consumed
+ while (!queue.isEmpty()) {
+ Thread.sleep(1);
+ }
+ }
+
+ latch.await();
+ }
+
+ @Test
+ public void growArray() throws Exception {
+ GrowableBatchedArrayBlockingQueue<Integer> queue = new
GrowableBatchedArrayBlockingQueue<>(4);
+
+ assertNull(queue.poll());
+
+ assertTrue(queue.offer(1));
+ assertTrue(queue.offer(2));
+ assertTrue(queue.offer(3));
+ assertTrue(queue.offer(4));
+ assertTrue(queue.offer(5));
+
+ assertEquals(5, queue.size());
+
+ queue.clear();
+ assertEquals(0, queue.size());
+
+ assertTrue(queue.offer(1, 1, TimeUnit.SECONDS));
+ assertTrue(queue.offer(2, 1, TimeUnit.SECONDS));
+ assertTrue(queue.offer(3, 1, TimeUnit.SECONDS));
+ assertEquals(3, queue.size());
+
+ List<Integer> list = new ArrayList<>();
+ queue.drainTo(list);
+ assertEquals(0, queue.size());
+
+ assertEquals(Lists.newArrayList(1, 2, 3), list);
+ }
+
+ @Test
+ public void growWithWrappedIndexes() throws Exception {
+ GrowableBatchedArrayBlockingQueue<Integer> queue = new
GrowableBatchedArrayBlockingQueue<>(4);
+
+ assertTrue(queue.offer(0));
+ assertTrue(queue.offer(1));
+ assertTrue(queue.offer(2));
+
+ // Advance the consumer index so that the queue content wraps around
the array
+ assertEquals(0, queue.poll().intValue());
+ assertEquals(1, queue.poll().intValue());
+
+ assertTrue(queue.offer(3));
+ assertTrue(queue.offer(4));
+ assertTrue(queue.offer(5));
+
+ // Next offer triggers the growth with wrapped content
+ assertTrue(queue.offer(6));
+ assertEquals(5, queue.size());
+
+ for (int i = 2; i <= 6; i++) {
+ assertEquals(i, queue.take().intValue());
+ }
+ assertEquals(0, queue.size());
+ }
+
+ @Test
+ public void putAllAndTakeAll() throws Exception {
+ GrowableBatchedArrayBlockingQueue<Integer> queue = new
GrowableBatchedArrayBlockingQueue<>(4);
+
+ Integer[] items = new Integer[100];
+ for (int i = 0; i < 100; i++) {
+ items[i] = i;
+ }
+
+ // Insert, in one shot, more items than the current capacity
+ queue.putAll(items, 0, 100);
+ assertEquals(100, queue.size());
+
+ Integer[] local = new Integer[100];
+ int n = queue.takeAll(local);
+ assertEquals(100, n);
+ for (int i = 0; i < 100; i++) {
+ assertEquals(i, local[i].intValue());
+ }
+ assertEquals(0, queue.size());
+ }
+
+ @Test
+ public void putAllWithWrappedIndexes() throws Exception {
+ GrowableBatchedArrayBlockingQueue<Integer> queue = new
GrowableBatchedArrayBlockingQueue<>(8);
+
+ // Advance the indexes so that the batch insert wraps around the array
+ for (int i = 0; i < 6; i++) {
+ queue.put(i);
+ }
+ for (int i = 0; i < 6; i++) {
+ assertEquals(i, queue.take().intValue());
+ }
+
+ Integer[] items = new Integer[4];
+ for (int i = 0; i < 4; i++) {
+ items[i] = 100 + i;
+ }
+
+ queue.putAll(items, 0, 4);
+ assertEquals(4, queue.size());
+
+ for (int i = 0; i < 4; i++) {
+ assertEquals(100 + i, queue.take().intValue());
+ }
+ }
+
+ @Test
+ public void takeAllBlocksUntilItemsAreAvailable() throws Exception {
+ GrowableBatchedArrayBlockingQueue<Integer> queue = new
GrowableBatchedArrayBlockingQueue<>();
+
+ CountDownLatch latch = new CountDownLatch(1);
+
+ new Thread(() -> {
+ try {
+ Integer[] local = new Integer[10];
+ int n = queue.takeAll(local);
+ assertEquals(1, n);
+ assertEquals(1, local[0].intValue());
+
+ latch.countDown();
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }).start();
+
+ // Make sure the background thread is waiting on takeAll
+ Thread.sleep(100);
+ queue.put(1);
+
+ latch.await();
+ }
+
+ @Test
+ public void pollAllTimeout() throws Exception {
+ GrowableBatchedArrayBlockingQueue<Integer> queue = new
GrowableBatchedArrayBlockingQueue<>();
+
+ Integer[] local = new Integer[10];
+ assertEquals(0, queue.pollAll(local, 1, TimeUnit.MILLISECONDS));
+
+ queue.put(1);
+ queue.put(2);
+ assertEquals(2, queue.pollAll(local, 1, TimeUnit.MILLISECONDS));
+ assertEquals(1, local[0].intValue());
+ assertEquals(2, local[1].intValue());
+ }
+
+ @Test
+ public void pollTimeout() throws Exception {
+ GrowableBatchedArrayBlockingQueue<Integer> queue = new
GrowableBatchedArrayBlockingQueue<>(4);
+
+ assertNull(queue.poll(1, TimeUnit.MILLISECONDS));
+
+ queue.put(1);
+ assertEquals(1, queue.poll(1, TimeUnit.MILLISECONDS).intValue());
+
+ // 0 timeout should not block
+ assertNull(queue.poll(0, TimeUnit.HOURS));
+
+ queue.put(2);
+ queue.put(3);
+ assertEquals(2, queue.poll(1, TimeUnit.HOURS).intValue());
+ assertEquals(3, queue.poll(1, TimeUnit.HOURS).intValue());
+ }
+
+ @Test(timeout = 10_000)
+ public void zeroInitialCapacityIsNormalized() throws Exception {
+ GrowableBatchedArrayBlockingQueue<Integer> queue = new
GrowableBatchedArrayBlockingQueue<>(0);
+
+ for (int i = 0; i < 10; i++) {
+ assertTrue(queue.offer(i));
+ }
+
+ for (int i = 0; i < 10; i++) {
+ assertEquals(i, queue.take().intValue());
+ }
+ }
+
+ @Test
+ public void rejectNullElements() {
+ GrowableBatchedArrayBlockingQueue<Integer> queue = new
GrowableBatchedArrayBlockingQueue<>();
+
+ assertThrows(NullPointerException.class, () -> queue.offer(null));
+ assertThrows(NullPointerException.class, () -> queue.put(null));
+ assertThrows(NullPointerException.class, () -> queue.add(null));
+ }
+
+ @Test
+ public void drainToValidation() {
+ GrowableBatchedArrayBlockingQueue<Integer> queue = new
GrowableBatchedArrayBlockingQueue<>();
+ queue.put(1);
+
+ assertThrows(NullPointerException.class, () -> queue.drainTo(null));
+ assertThrows(IllegalArgumentException.class, () ->
queue.drainTo(queue));
+
+ // A non-positive maxElements does not drain and does not corrupt the
queue state
+ List<Integer> list = new ArrayList<>();
+ assertEquals(0, queue.drainTo(list, -1));
+ assertEquals(0, queue.drainTo(list, 0));
+ assertEquals(1, queue.size());
+ assertEquals(1, queue.poll().intValue());
+ }
+
+ @Test
+ public void drainToIsExceptionSafe() {
+ GrowableBatchedArrayBlockingQueue<Integer> queue = new
GrowableBatchedArrayBlockingQueue<>();
+ for (int i = 0; i < 5; i++) {
+ queue.put(i);
+ }
+
+ List<Integer> target = new ArrayList<Integer>() {
+ @Override
+ public boolean add(Integer item) {
+ if (size() == 2) {
+ throw new RuntimeException("Rejecting item");
+ }
+ return super.add(item);
+ }
+ };
+
+ assertThrows(RuntimeException.class, () -> queue.drainTo(target));
+
+ // The two transferred items are committed, the rest are still in the
queue
+ assertEquals(Lists.newArrayList(0, 1), target);
+ assertEquals(3, queue.size());
+ for (int i = 2; i < 5; i++) {
+ assertEquals(i, queue.poll().intValue());
+ }
+ }
+
+ @Test(timeout = 10_000)
+ public void takeAllWithEmptyArrayDoesNotBlock() throws Exception {
+ GrowableBatchedArrayBlockingQueue<Integer> queue = new
GrowableBatchedArrayBlockingQueue<>();
+
+ // Even with an empty queue, a zero-length destination array returns
immediately
+ assertEquals(0, queue.takeAll(new Integer[0]));
+ assertEquals(0, queue.pollAll(new Integer[0], 1, TimeUnit.HOURS));
+ }
+
+ @Test
+ public void putAllArgumentValidation() {
+ GrowableBatchedArrayBlockingQueue<Integer> queue = new
GrowableBatchedArrayBlockingQueue<>();
+
+ assertThrows(NullPointerException.class, () -> queue.putAll(null, 0,
1));
+
+ Integer[] items = { 1, 2, 3 };
+ assertThrows(IndexOutOfBoundsException.class, () ->
queue.putAll(items, -1, 2));
+ assertThrows(IndexOutOfBoundsException.class, () ->
queue.putAll(items, 0, 4));
+ assertThrows(IndexOutOfBoundsException.class, () ->
queue.putAll(items, 2, 2));
+
+ assertEquals(0, queue.size());
+ }
+}
diff --git
a/microbenchmarks/src/main/java/org/apache/bookkeeper/common/SingleThreadExecutorQueueBenchmark.java
b/microbenchmarks/src/main/java/org/apache/bookkeeper/common/SingleThreadExecutorQueueBenchmark.java
new file mode 100644
index 0000000000..335a7b7303
--- /dev/null
+++
b/microbenchmarks/src/main/java/org/apache/bookkeeper/common/SingleThreadExecutorQueueBenchmark.java
@@ -0,0 +1,139 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.bookkeeper.common;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import lombok.SneakyThrows;
+import org.apache.bookkeeper.common.collections.BatchedArrayBlockingQueue;
+import org.apache.bookkeeper.common.collections.BatchedBlockingQueue;
+import
org.apache.bookkeeper.common.collections.GrowableBatchedArrayBlockingQueue;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+
+/**
+ * Compares the queue options for {@code SingleThreadExecutor}: the JDK {@code
ArrayBlockingQueue}
+ * drained with {@code drainTo}, and the batched queues ({@code
BatchedArrayBlockingQueue} and
+ * {@code GrowableBatchedArrayBlockingQueue}) drained with {@code takeAll}.
+ *
+ * <p>The consumer threads mirror the drain patterns of {@code
SingleThreadExecutor.run()}.
+ */
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+@BenchmarkMode(Mode.Throughput)
+@Threads(8)
+@Fork(1)
+@Warmup(iterations = 2, time = 5, timeUnit = TimeUnit.SECONDS)
+@Measurement(iterations = 3, time = 5, timeUnit = TimeUnit.SECONDS)
+public class SingleThreadExecutorQueueBenchmark {
+
+ private static final int QUEUE_SIZE = 100_000;
+
+ /**
+ * State holder of the test.
+ */
+ @State(Scope.Benchmark)
+ public static class TestState {
+
+ private final ArrayBlockingQueue<Integer> arrayBlockingQueue = new
ArrayBlockingQueue<>(QUEUE_SIZE);
+
+ private final BatchedArrayBlockingQueue<Integer> batchedQueue = new
BatchedArrayBlockingQueue<>(QUEUE_SIZE);
+
+ private final GrowableBatchedArrayBlockingQueue<Integer>
growableBatchedQueue =
+ new GrowableBatchedArrayBlockingQueue<>();
+
+ private final ExecutorService executor =
Executors.newCachedThreadPool();
+
+ @Setup(Level.Trial)
+ public void setup() {
+ executor.execute(() -> consumeWithDrainTo(arrayBlockingQueue));
+ executor.execute(() -> consumeWithTakeAll(batchedQueue));
+ executor.execute(() -> consumeWithTakeAll(growableBatchedQueue));
+ }
+
+ @SneakyThrows
+ private void consumeWithDrainTo(BlockingQueue<Integer> queue) {
+ List<Integer> localTasks = new ArrayList<>();
+
+ try {
+ while (true) {
+ int n = queue.drainTo(localTasks);
+ if (n > 0) {
+ localTasks.clear();
+ } else {
+ queue.take();
+ }
+ }
+ } catch (InterruptedException ie) {
+ }
+ }
+
+ @SneakyThrows
+ private void consumeWithTakeAll(BatchedBlockingQueue<Integer> queue) {
+ Integer[] localTasks = new Integer[20_000];
+
+ try {
+ while (true) {
+ queue.takeAll(localTasks);
+ }
+ } catch (InterruptedException ie) {
+ }
+ }
+
+ @TearDown(Level.Trial)
+ public void teardown() {
+ executor.shutdownNow();
+ }
+
+ @TearDown(Level.Iteration)
+ public void drainBacklog() throws InterruptedException {
+ Thread.sleep(1_000);
+ }
+ }
+
+ @Benchmark
+ public void arrayBlockingQueue(TestState s) throws Exception {
+ s.arrayBlockingQueue.put(1);
+ }
+
+ @Benchmark
+ public void batchedArrayBlockingQueue(TestState s) throws Exception {
+ s.batchedQueue.put(1);
+ }
+
+ @Benchmark
+ public void growableBatchedArrayBlockingQueue(TestState s) throws
Exception {
+ s.growableBatchedQueue.put(1);
+ }
+}
