diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/QuorumCnxManager.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/QuorumCnxManager.java index 23274459307..e37986d4e78 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/QuorumCnxManager.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/QuorumCnxManager.java @@ -36,9 +36,8 @@ import java.util.Enumeration; import java.util.HashSet; import java.util.Map; -import java.util.NoSuchElementException; import java.util.Set; -import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.SynchronousQueue; import java.util.concurrent.ThreadFactory; @@ -55,6 +54,7 @@ import org.apache.zookeeper.server.quorum.auth.QuorumAuthServer; import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier; import org.apache.zookeeper.server.util.ConfigUtils; +import org.apache.zookeeper.util.CircularBlockingQueue; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -137,17 +137,13 @@ public class QuorumCnxManager { * Mapping from Peer to Thread number */ final ConcurrentHashMap senderWorkerMap; - final ConcurrentHashMap> queueSendMap; + final ConcurrentHashMap> queueSendMap; final ConcurrentHashMap lastMessageSent; /* * Reception queue */ - public final ArrayBlockingQueue recvQueue; - /* - * Object to synchronize access to recvQueue - */ - private final Object recvQLock = new Object(); + public final BlockingQueue recvQueue; /* * Shutdown flag @@ -253,10 +249,10 @@ public static InitialMessage parse(Long protocolVersion, DataInputStream din) th } public QuorumCnxManager(QuorumPeer self, final long mySid, Map view, QuorumAuthServer authServer, QuorumAuthLearner authLearner, int socketTimeout, boolean listenOnAllIPs, int quorumCnxnThreadsSize, boolean quorumSaslAuthEnabled) { - this.recvQueue = new ArrayBlockingQueue(RECV_CAPACITY); - this.queueSendMap = new ConcurrentHashMap>(); - this.senderWorkerMap = new ConcurrentHashMap(); - this.lastMessageSent = new ConcurrentHashMap(); + this.recvQueue = new CircularBlockingQueue<>(RECV_CAPACITY); + this.queueSendMap = new ConcurrentHashMap<>(); + this.senderWorkerMap = new ConcurrentHashMap<>(); + this.lastMessageSent = new ConcurrentHashMap<>(); String cnxToValue = System.getProperty("zookeeper.cnxTimeout"); if (cnxToValue != null) { @@ -438,7 +434,8 @@ private boolean startConnection(Socket sock, Long sid) throws IOException { } senderWorkerMap.put(sid, sw); - queueSendMap.putIfAbsent(sid, new ArrayBlockingQueue(SEND_CAPACITY)); + + queueSendMap.putIfAbsent(sid, new CircularBlockingQueue<>(SEND_CAPACITY)); sw.start(); rw.start(); @@ -573,7 +570,7 @@ private void handleConnection(Socket sock, DataInputStream din) throws IOExcepti senderWorkerMap.put(sid, sw); - queueSendMap.putIfAbsent(sid, new ArrayBlockingQueue(SEND_CAPACITY)); + queueSendMap.putIfAbsent(sid, new CircularBlockingQueue<>(SEND_CAPACITY)); sw.start(); rw.start(); @@ -598,10 +595,9 @@ public void toSend(Long sid, ByteBuffer b) { /* * Start a new connection if doesn't have one already. */ - ArrayBlockingQueue bq = queueSendMap.computeIfAbsent(sid, serverId -> new ArrayBlockingQueue<>(SEND_CAPACITY)); + BlockingQueue bq = queueSendMap.computeIfAbsent(sid, serverId -> new CircularBlockingQueue<>(SEND_CAPACITY)); addToSendQueue(bq, b); connectOne(sid); - } } @@ -724,9 +720,10 @@ public void connectAll() { * Check if all queues are empty, indicating that all messages have been delivered. */ boolean haveDelivered() { - for (ArrayBlockingQueue queue : queueSendMap.values()) { - LOG.debug("Queue size: {}", queue.size()); - if (queue.size() == 0) { + for (BlockingQueue queue : queueSendMap.values()) { + final int queueSize = queue.size(); + LOG.debug("Queue size: {}", queueSize); + if (queueSize == 0) { return true; } } @@ -1085,7 +1082,7 @@ public void run() { * message than that stored in lastMessage. To avoid sending * stale message, we should send the message in the send queue. */ - ArrayBlockingQueue bq = queueSendMap.get(sid); + BlockingQueue bq = queueSendMap.get(sid); if (bq == null || isSendQueueEmpty(bq)) { ByteBuffer b = lastMessageSent.get(sid); if (b != null) { @@ -1103,7 +1100,7 @@ public void run() { ByteBuffer b = null; try { - ArrayBlockingQueue bq = queueSendMap.get(sid); + BlockingQueue bq = queueSendMap.get(sid); if (bq != null) { b = pollSendQueue(bq, 1000, TimeUnit.MILLISECONDS); } else { @@ -1216,37 +1213,19 @@ public void run() { } /** - * Inserts an element in the specified queue. If the Queue is full, this - * method removes an element from the head of the Queue and then inserts - * the element at the tail. It can happen that an element is removed - * by another thread in {@link SendWorker#run() } - * method before this method attempts to remove an element from the queue. - * This will cause {@link ArrayBlockingQueue#remove() remove} to throw an - * exception, which is safe to ignore. - * - * Unlike {@link #addToRecvQueue(Message) addToRecvQueue} this method does - * not need to be synchronized since there is only one thread that inserts - * an element in the queue and another thread that reads from the queue. + * Inserts an element in the provided {@link BlockingQueue}. This method + * assumes that if the Queue is full, an element from the head of the Queue is + * removed and the new item is inserted at the tail of the queue. This is done + * to prevent a thread from blocking while inserting an element in the queue. * - * @param queue - * Reference to the Queue - * @param buffer - * Reference to the buffer to be inserted in the queue + * @param queue Reference to the Queue + * @param buffer Reference to the buffer to be inserted in the queue */ - private void addToSendQueue(ArrayBlockingQueue queue, ByteBuffer buffer) { - if (queue.remainingCapacity() == 0) { - try { - queue.remove(); - } catch (NoSuchElementException ne) { - // element could be removed by poll() - LOG.debug("Trying to remove from an empty Queue. Ignoring exception.", ne); - } - } - try { - queue.add(buffer); - } catch (IllegalStateException ie) { - // This should never happen - LOG.error("Unable to insert an element in the queue ", ie); + private void addToSendQueue(final BlockingQueue queue, + final ByteBuffer buffer) { + final boolean success = queue.offer(buffer); + if (!success) { + throw new RuntimeException("Could not insert into receive queue"); } } @@ -1257,7 +1236,7 @@ private void addToSendQueue(ArrayBlockingQueue queue, ByteBuffer buf * @return * true if the specified queue is empty */ - private boolean isSendQueueEmpty(ArrayBlockingQueue queue) { + private boolean isSendQueueEmpty(final BlockingQueue queue) { return queue.isEmpty(); } @@ -1266,49 +1245,25 @@ private boolean isSendQueueEmpty(ArrayBlockingQueue queue) { * waiting up to the specified wait time if necessary for an element to * become available. * - * {@link ArrayBlockingQueue#poll(long, java.util.concurrent.TimeUnit)} + * {@link BlockingQueue#poll(long, java.util.concurrent.TimeUnit)} */ - private ByteBuffer pollSendQueue(ArrayBlockingQueue queue, long timeout, TimeUnit unit) throws InterruptedException { - return queue.poll(timeout, unit); + private ByteBuffer pollSendQueue(final BlockingQueue queue, + final long timeout, final TimeUnit unit) throws InterruptedException { + return queue.poll(timeout, unit); } /** * Inserts an element in the {@link #recvQueue}. If the Queue is full, this - * methods removes an element from the head of the Queue and then inserts - * the element at the tail of the queue. - * - * This method is synchronized to achieve fairness between two threads that - * are trying to insert an element in the queue. Each thread checks if the - * queue is full, then removes the element at the head of the queue, and - * then inserts an element at the tail. This three-step process is done to - * prevent a thread from blocking while inserting an element in the queue. - * If we do not synchronize the call to this method, then a thread can grab - * a slot in the queue created by the second thread. This can cause the call - * to insert by the second thread to fail. - * Note that synchronizing this method does not block another thread - * from polling the queue since that synchronization is provided by the - * queue itself. + * methods removes an element from the head of the Queue and then inserts the + * element at the tail of the queue. * - * @param msg - * Reference to the message to be inserted in the queue + * @param msg Reference to the message to be inserted in the queue */ - public void addToRecvQueue(Message msg) { - synchronized (recvQLock) { - if (recvQueue.remainingCapacity() == 0) { - try { - recvQueue.remove(); - } catch (NoSuchElementException ne) { - // element could be removed by poll() - LOG.debug("Trying to remove from an empty recvQueue. Ignoring exception.", ne); - } - } - try { - recvQueue.add(msg); - } catch (IllegalStateException ie) { - // This should never happen - LOG.error("Unable to insert element in the recvQueue ", ie); - } - } + public void addToRecvQueue(final Message msg) { + final boolean success = this.recvQueue.offer(msg); + if (!success) { + throw new RuntimeException("Could not insert into receive queue"); + } } /** @@ -1316,10 +1271,11 @@ public void addToRecvQueue(Message msg) { * waiting up to the specified wait time if necessary for an element to * become available. * - * {@link ArrayBlockingQueue#poll(long, java.util.concurrent.TimeUnit)} + * {@link BlockingQueue#poll(long, java.util.concurrent.TimeUnit)} */ - public Message pollRecvQueue(long timeout, TimeUnit unit) throws InterruptedException { - return recvQueue.poll(timeout, unit); + public Message pollRecvQueue(final long timeout, final TimeUnit unit) + throws InterruptedException { + return this.recvQueue.poll(timeout, unit); } public boolean connectedToPeer(long peerSid) { diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/util/CircularBlockingQueue.java b/zookeeper-server/src/main/java/org/apache/zookeeper/util/CircularBlockingQueue.java new file mode 100644 index 00000000000..cbacb658ad2 --- /dev/null +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/util/CircularBlockingQueue.java @@ -0,0 +1,277 @@ +/** + * 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.zookeeper.util; + +import java.util.ArrayDeque; +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.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A bounded blocking queue backed by an array. This queue orders elements FIFO + * (first-in-first-out). The head of the queue is that element that has been on + * the queue the longest time. The tail of the queue is that element that has + * been on the queue the shortest time. New elements are inserted at the tail of + * the queue, and the queue retrieval operations obtain elements at the head of + * the queue. If the queue is full, the head of the queue (the oldest element) + * will be removed to make room for the newest element. + */ +public class CircularBlockingQueue implements BlockingQueue { + + private static final Logger LOG = LoggerFactory.getLogger(CircularBlockingQueue.class); + + /** Main lock guarding all access */ + private final ReentrantLock lock; + + /** Condition for waiting takes */ + private final Condition notEmpty; + + /** The array-backed queue */ + private final ArrayDeque queue; + + private final int maxSize; + + private long droppedCount; + + public CircularBlockingQueue(int queueSize) { + this.queue = new ArrayDeque<>(queueSize); + this.maxSize = queueSize; + + this.lock = new ReentrantLock(); + this.notEmpty = this.lock.newCondition(); + this.droppedCount = 0L; + } + + /** + * This method differs from {@link BlockingQueue#offer(Object)} in that it + * will remove the oldest queued element (the element at the front of the + * queue) in order to make room for any new elements if the queue is full. + * + * @param e the element to add + * @return true since it will make room for any new elements if required + */ + @Override + public boolean offer(E e) { + Objects.requireNonNull(e); + final ReentrantLock lock = this.lock; + lock.lock(); + try { + if (this.queue.size() == this.maxSize) { + final E discard = this.queue.remove(); + this.droppedCount++; + LOG.debug("Queue is full. Discarding oldest element [count={}]: {}", + this.droppedCount, discard); + } + this.queue.add(e); + this.notEmpty.signal(); + } finally { + lock.unlock(); + } + return true; + } + + @Override + public E poll(long timeout, TimeUnit unit) throws InterruptedException { + long nanos = unit.toNanos(timeout); + final ReentrantLock lock = this.lock; + lock.lockInterruptibly(); + try { + while (this.queue.isEmpty()) { + if (nanos <= 0) { + return null; + } + nanos = this.notEmpty.awaitNanos(nanos); + } + return this.queue.poll(); + } finally { + lock.unlock(); + } + } + + @Override + public E take() throws InterruptedException { + final ReentrantLock lock = this.lock; + lock.lockInterruptibly(); + try { + while (this.queue.isEmpty()) { + this.notEmpty.await(); + } + return this.queue.poll(); + } finally { + lock.unlock(); + } + } + + @Override + public boolean isEmpty() { + final ReentrantLock lock = this.lock; + lock.lock(); + try { + return this.queue.isEmpty(); + } finally { + lock.unlock(); + } + } + + @Override + public int size() { + final ReentrantLock lock = this.lock; + lock.lock(); + try { + return this.queue.size(); + } finally { + lock.unlock(); + } + } + + /** + * Returns the number of elements that were dropped from the queue because the + * queue was full when a new element was offered. + * + * @return The number of elements dropped (lost) from the queue + */ + public long getDroppedCount() { + return this.droppedCount; + } + + /** + * For testing purposes only. + * + * @return True if a thread is blocked waiting for a new element to be offered + * to the queue + */ + boolean isConsumerThreadBlocked() { + final ReentrantLock lock = this.lock; + lock.lock(); + try { + return lock.getWaitQueueLength(this.notEmpty) > 0; + } finally { + lock.unlock(); + } + } + + @Override + public int drainTo(Collection c) { + throw new UnsupportedOperationException(); + } + + + @Override + public E poll() { + throw new UnsupportedOperationException(); + } + + @Override + public E element() { + throw new UnsupportedOperationException(); + } + + @Override + public E peek() { + throw new UnsupportedOperationException(); + } + + @Override + public E remove() { + throw new UnsupportedOperationException(); + } + + @Override + public boolean addAll(Collection arg0) { + throw new UnsupportedOperationException(); + } + + @Override + public void clear() { + throw new UnsupportedOperationException(); + } + + @Override + public boolean containsAll(Collection arg0) { + throw new UnsupportedOperationException(); + } + + @Override + public Iterator iterator() { + throw new UnsupportedOperationException(); + } + + @Override + public boolean removeAll(Collection arg0) { + throw new UnsupportedOperationException(); + } + + @Override + public boolean retainAll(Collection arg0) { + throw new UnsupportedOperationException(); + } + + @Override + public Object[] toArray() { + throw new UnsupportedOperationException(); + } + + @Override + public T[] toArray(T[] arg0) { + throw new UnsupportedOperationException(); + } + + @Override + public boolean add(E e) { + throw new UnsupportedOperationException(); + } + + @Override + public boolean contains(Object o) { + throw new UnsupportedOperationException(); + } + + @Override + public int drainTo(Collection c, int maxElements) { + throw new UnsupportedOperationException(); + } + + @Override + public boolean offer(E e, long timeout, TimeUnit unit) + throws InterruptedException { + throw new UnsupportedOperationException(); + } + + @Override + public void put(E e) throws InterruptedException { + throw new UnsupportedOperationException(); + } + + @Override + public int remainingCapacity() { + throw new UnsupportedOperationException(); + } + + @Override + public boolean remove(Object o) { + throw new UnsupportedOperationException(); + } + +} diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/util/TestCircularBlockingQueue.java b/zookeeper-server/src/test/java/org/apache/zookeeper/util/TestCircularBlockingQueue.java new file mode 100644 index 00000000000..ac24d2e5b4d --- /dev/null +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/util/TestCircularBlockingQueue.java @@ -0,0 +1,76 @@ +/** + * 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.zookeeper.util; + +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import org.junit.Assert; +import org.junit.Test; + +public class TestCircularBlockingQueue { + + @Test + public void testCircularBlockingQueue() throws InterruptedException { + final CircularBlockingQueue testQueue = + new CircularBlockingQueue<>(2); + + testQueue.offer(1); + testQueue.offer(2); + testQueue.offer(3); + + Assert.assertEquals(2, testQueue.size()); + + Assert.assertEquals(2, testQueue.take().intValue()); + Assert.assertEquals(3, testQueue.take().intValue()); + + Assert.assertEquals(1L, testQueue.getDroppedCount()); + Assert.assertEquals(0, testQueue.size()); + Assert.assertEquals(true, testQueue.isEmpty()); + } + + @Test(timeout = 10000L) + public void testCircularBlockingQueueTakeBlock() + throws InterruptedException, ExecutionException { + + final CircularBlockingQueue testQueue = new CircularBlockingQueue<>(2); + + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future testTake = executor.submit(() -> { + return testQueue.take(); + }); + + // Allow the other thread to get into position; waiting for item to be + // inserted + while (!testQueue.isConsumerThreadBlocked()) { + Thread.sleep(50L); + } + + testQueue.offer(10); + + Integer result = testTake.get(); + Assert.assertEquals(10, result.intValue()); + } finally { + executor.shutdown(); + } + } + +}