From ce96b707132fea5a546d06ee8f97c408955ca76f Mon Sep 17 00:00:00 2001 From: Beluga Behr Date: Thu, 28 Mar 2019 16:49:03 -0400 Subject: [PATCH 1/8] ZOOKEEPER-3340: Improve Queue Usage in QuorumCnxManager.java --- .../server/quorum/QuorumCnxManager.java | 73 +++++++------------ 1 file changed, 27 insertions(+), 46 deletions(-) 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..bfd18442805 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,9 @@ 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; @@ -143,11 +143,7 @@ public class QuorumCnxManager { /* * Reception queue */ - public final ArrayBlockingQueue recvQueue; - /* - * Object to synchronize access to recvQueue - */ - private final Object recvQLock = new Object(); + public final BlockingQueue recvQueue; /* * Shutdown flag @@ -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 ArrayBlockingQueue<>(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 ArrayBlockingQueue<>(SEND_CAPACITY)); sw.start(); rw.start(); @@ -601,7 +598,6 @@ public void toSend(Long sid, ByteBuffer b) { ArrayBlockingQueue bq = queueSendMap.computeIfAbsent(sid, serverId -> new ArrayBlockingQueue<>(SEND_CAPACITY)); addToSendQueue(bq, b); connectOne(sid); - } } @@ -1085,7 +1081,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 +1099,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 { @@ -1233,20 +1229,12 @@ public void run() { * @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) { + queue.poll(); + queue.offer(buffer); } } @@ -1257,7 +1245,7 @@ private void addToSendQueue(ArrayBlockingQueue queue, ByteBuffer buf * @return * true if the specified queue is empty */ - private boolean isSendQueueEmpty(ArrayBlockingQueue queue) { + private boolean isSendQueueEmpty(BlockingQueue queue) { return queue.isEmpty(); } @@ -1266,10 +1254,11 @@ 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); } /** @@ -1292,21 +1281,12 @@ private ByteBuffer pollSendQueue(ArrayBlockingQueue queue, long time * @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) { + synchronized (this.recvQueue) { + final boolean success = this.recvQueue.offer(msg); + if (!success) { + this.recvQueue.poll(); + this.recvQueue.offer(msg); } } } @@ -1318,8 +1298,9 @@ public void addToRecvQueue(Message msg) { * * {@link ArrayBlockingQueue#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) { From 2793db2f1a4b1cb5f2b103f842a7b397f57db0a6 Mon Sep 17 00:00:00 2001 From: Beluga Behr Date: Fri, 29 Mar 2019 09:10:19 -0400 Subject: [PATCH 2/8] Added additional logging --- .../org/apache/zookeeper/server/quorum/QuorumCnxManager.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) 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 bfd18442805..38b78cad68e 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 @@ -1233,6 +1233,7 @@ private void addToSendQueue(final BlockingQueue queue, final ByteBuffer buffer) { final boolean success = queue.offer(buffer); if (!success) { + LOG.debug("Could not insert buffer into queue. Discarding one."); queue.poll(); queue.offer(buffer); } @@ -1245,7 +1246,7 @@ private void addToSendQueue(final BlockingQueue queue, * @return * true if the specified queue is empty */ - private boolean isSendQueueEmpty(BlockingQueue queue) { + private boolean isSendQueueEmpty(final BlockingQueue queue) { return queue.isEmpty(); } @@ -1285,6 +1286,7 @@ public void addToRecvQueue(final Message msg) { synchronized (this.recvQueue) { final boolean success = this.recvQueue.offer(msg); if (!success) { + LOG.debug("Could not insert buffer into recv queue. Discarding one."); this.recvQueue.poll(); this.recvQueue.offer(msg); } From d8877803b7921a1c34b43e2a3c15b630558955e2 Mon Sep 17 00:00:00 2001 From: Beluga Behr Date: Fri, 29 Mar 2019 14:36:32 -0400 Subject: [PATCH 3/8] Introduce new class CircularBlockingQueue --- .../zookeeper/util/CircularBlockingQueue.java | 246 ++++++++++++++++++ .../util/TestCircularBlockingQueue.java | 70 +++++ 2 files changed, 316 insertions(+) create mode 100644 zookeeper-server/src/main/java/org/apache/zookeeper/util/CircularBlockingQueue.java create mode 100644 zookeeper-server/src/test/java/org/apache/zookeeper/server/util/TestCircularBlockingQueue.java 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..828c895becf --- /dev/null +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/util/CircularBlockingQueue.java @@ -0,0 +1,246 @@ +/** + * 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; + + public CircularBlockingQueue(int queueSize) { + this.queue = new ArrayDeque<>(queueSize); + this.maxSize = queueSize; + + this.lock = new ReentrantLock(); + this.notEmpty = this.lock.newCondition(); + } + + /** + * 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(); + LOG.debug("Queue if full. Discarding oldest element: {}", 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(); + } + } + + @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/server/util/TestCircularBlockingQueue.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/util/TestCircularBlockingQueue.java new file mode 100644 index 00000000000..88c124d35fd --- /dev/null +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/server/util/TestCircularBlockingQueue.java @@ -0,0 +1,70 @@ +/** + * 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.server.util; + +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; + +import org.apache.zookeeper.util.CircularBlockingQueue; +import org.junit.Assert; +import org.junit.Test; + +public class TestCircularBlockingQueue { + + @Test + public void testCircularBlockingQueue() throws InterruptedException { + final BlockingQueue 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(0, testQueue.size()); + Assert.assertEquals(true, testQueue.isEmpty()); + } + + @Test(timeout = 10000L) + public void testCircularBlockingQueueTakeBlock() + throws InterruptedException, ExecutionException { + + final BlockingQueue testQueue = new CircularBlockingQueue<>(2); + + ExecutorService executor = Executors.newSingleThreadExecutor(); + + Future testTake = executor.submit(() -> { + return testQueue.take(); + }); + + // Allow the other thread to get into position; waiting for item to be inserted + Thread.sleep(2000L); + + testQueue.offer(10); + + Integer result = testTake.get(); + Assert.assertEquals(10, result.intValue()); + } + +} From c63ab852c954e21ed0d70184e2acf881ff85d22b Mon Sep 17 00:00:00 2001 From: Beluga Behr Date: Mon, 8 Apr 2019 10:12:17 -0400 Subject: [PATCH 4/8] Added updates from code review --- .../zookeeper/util/CircularBlockingQueue.java | 34 ++++++++++++++++++- .../util/TestCircularBlockingQueue.java | 15 ++++---- 2 files changed, 42 insertions(+), 7 deletions(-) rename zookeeper-server/src/test/java/org/apache/zookeeper/{server => }/util/TestCircularBlockingQueue.java (84%) 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 index 828c895becf..874c4c91bbb 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/util/CircularBlockingQueue.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/util/CircularBlockingQueue.java @@ -54,12 +54,15 @@ public class CircularBlockingQueue implements BlockingQueue { 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; } /** @@ -78,7 +81,9 @@ public boolean offer(E e) { try { if (this.queue.size() == this.maxSize) { final E discard = this.queue.remove(); - LOG.debug("Queue if full. Discarding oldest element: {}", discard); + this.droppedCount++; + LOG.debug("Queue is full. Discarding oldest element [count={}]: {}", + this.droppedCount, discard); } this.queue.add(e); this.notEmpty.signal(); @@ -142,11 +147,38 @@ public int size() { } } + /** + * 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(); diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/util/TestCircularBlockingQueue.java b/zookeeper-server/src/test/java/org/apache/zookeeper/util/TestCircularBlockingQueue.java similarity index 84% rename from zookeeper-server/src/test/java/org/apache/zookeeper/server/util/TestCircularBlockingQueue.java rename to zookeeper-server/src/test/java/org/apache/zookeeper/util/TestCircularBlockingQueue.java index 88c124d35fd..87401f4cbd1 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/server/util/TestCircularBlockingQueue.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/util/TestCircularBlockingQueue.java @@ -16,15 +16,13 @@ * limitations under the License. */ -package org.apache.zookeeper.server.util; +package org.apache.zookeeper.util; -import java.util.concurrent.BlockingQueue; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; -import org.apache.zookeeper.util.CircularBlockingQueue; import org.junit.Assert; import org.junit.Test; @@ -32,7 +30,9 @@ public class TestCircularBlockingQueue { @Test public void testCircularBlockingQueue() throws InterruptedException { - final BlockingQueue testQueue = new CircularBlockingQueue<>(2); + final CircularBlockingQueue testQueue = + new CircularBlockingQueue<>(2); + testQueue.offer(1); testQueue.offer(2); testQueue.offer(3); @@ -42,6 +42,7 @@ public void testCircularBlockingQueue() throws InterruptedException { 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()); } @@ -50,7 +51,7 @@ public void testCircularBlockingQueue() throws InterruptedException { public void testCircularBlockingQueueTakeBlock() throws InterruptedException, ExecutionException { - final BlockingQueue testQueue = new CircularBlockingQueue<>(2); + final CircularBlockingQueue testQueue = new CircularBlockingQueue<>(2); ExecutorService executor = Executors.newSingleThreadExecutor(); @@ -59,7 +60,9 @@ public void testCircularBlockingQueueTakeBlock() }); // Allow the other thread to get into position; waiting for item to be inserted - Thread.sleep(2000L); + while (!testQueue.isConsumerThreadBlocked()) { + Thread.sleep(50L); + } testQueue.offer(10); From b3ac3cc44bea364e5a03a6e3867a085fee7eb20b Mon Sep 17 00:00:00 2001 From: David Mollitor Date: Thu, 19 Sep 2019 10:28:30 -0400 Subject: [PATCH 5/8] Rebased to master --- .../server/quorum/QuorumCnxManager.java | 83 +++++++------------ 1 file changed, 30 insertions(+), 53 deletions(-) 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 38b78cad68e..3ca23e9f058 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 @@ -19,6 +19,7 @@ package org.apache.zookeeper.server.quorum; import static org.apache.zookeeper.common.NetUtils.formatInetAddr; + import java.io.BufferedInputStream; import java.io.BufferedOutputStream; import java.io.DataInputStream; @@ -46,7 +47,9 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; + import javax.net.ssl.SSLSocket; + import org.apache.zookeeper.common.X509Exception; import org.apache.zookeeper.server.ExitCode; import org.apache.zookeeper.server.ZooKeeperThread; @@ -55,6 +58,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,7 +141,7 @@ public class QuorumCnxManager { * Mapping from Peer to Thread number */ final ConcurrentHashMap senderWorkerMap; - final ConcurrentHashMap> queueSendMap; + final ConcurrentHashMap> queueSendMap; final ConcurrentHashMap lastMessageSent; /* @@ -249,10 +253,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) { @@ -435,7 +439,7 @@ 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(); @@ -570,7 +574,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(); @@ -595,7 +599,7 @@ 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); } @@ -720,9 +724,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; } } @@ -1212,30 +1217,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(final BlockingQueue queue, final ByteBuffer buffer) { final boolean success = queue.offer(buffer); if (!success) { - LOG.debug("Could not insert buffer into queue. Discarding one."); - queue.poll(); - queue.offer(buffer); + throw new RuntimeException("Could not insert into receive queue"); } } @@ -1264,33 +1258,16 @@ private ByteBuffer pollSendQueue(final BlockingQueue queue, /** * 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(final Message msg) { - synchronized (this.recvQueue) { - final boolean success = this.recvQueue.offer(msg); - if (!success) { - LOG.debug("Could not insert buffer into recv queue. Discarding one."); - this.recvQueue.poll(); - this.recvQueue.offer(msg); - } - } + final boolean success = this.recvQueue.offer(msg); + if (!success) { + throw new RuntimeException("Could not insert into receive queue"); + } } /** From f5ea2b6b19b23f4e2375e5528f9b8c81d1f3c8f2 Mon Sep 17 00:00:00 2001 From: David Mollitor Date: Thu, 19 Sep 2019 10:45:19 -0400 Subject: [PATCH 6/8] Updated to fix checkstyle ImportOrder: Extra separation --- .../org/apache/zookeeper/server/quorum/QuorumCnxManager.java | 3 --- .../java/org/apache/zookeeper/util/CircularBlockingQueue.java | 1 - .../org/apache/zookeeper/util/TestCircularBlockingQueue.java | 1 - 3 files changed, 5 deletions(-) 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 3ca23e9f058..2f1bd5cd45c 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 @@ -19,7 +19,6 @@ package org.apache.zookeeper.server.quorum; import static org.apache.zookeeper.common.NetUtils.formatInetAddr; - import java.io.BufferedInputStream; import java.io.BufferedOutputStream; import java.io.DataInputStream; @@ -47,9 +46,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; - import javax.net.ssl.SSLSocket; - import org.apache.zookeeper.common.X509Exception; import org.apache.zookeeper.server.ExitCode; import org.apache.zookeeper.server.ZooKeeperThread; 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 index 874c4c91bbb..cbacb658ad2 100644 --- a/zookeeper-server/src/main/java/org/apache/zookeeper/util/CircularBlockingQueue.java +++ b/zookeeper-server/src/main/java/org/apache/zookeeper/util/CircularBlockingQueue.java @@ -26,7 +26,6 @@ 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; 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 index 87401f4cbd1..4243263f4cc 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/util/TestCircularBlockingQueue.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/util/TestCircularBlockingQueue.java @@ -22,7 +22,6 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; - import org.junit.Assert; import org.junit.Test; From 38d7e3f05336f7cf3dee97f78d210da06349f7b0 Mon Sep 17 00:00:00 2001 From: David Mollitor Date: Mon, 7 Oct 2019 11:39:29 -0400 Subject: [PATCH 7/8] Shutdown ExecutorService in test --- .../util/TestCircularBlockingQueue.java | 30 +++++++++++-------- 1 file changed, 17 insertions(+), 13 deletions(-) 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 index 4243263f4cc..ac24d2e5b4d 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/util/TestCircularBlockingQueue.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/util/TestCircularBlockingQueue.java @@ -53,20 +53,24 @@ public void testCircularBlockingQueueTakeBlock() final CircularBlockingQueue testQueue = new CircularBlockingQueue<>(2); ExecutorService executor = Executors.newSingleThreadExecutor(); - - 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); + 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(); } - - testQueue.offer(10); - - Integer result = testTake.get(); - Assert.assertEquals(10, result.intValue()); } } From 7b74dac278f2261218ac1818f51d721f28096952 Mon Sep 17 00:00:00 2001 From: David Mollitor Date: Thu, 7 Nov 2019 12:31:56 -0500 Subject: [PATCH 8/8] Changed comment from conrete ArrayBlockingQueue to BlockingQueue --- .../org/apache/zookeeper/server/quorum/QuorumCnxManager.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) 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 2f1bd5cd45c..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 @@ -37,7 +37,6 @@ import java.util.HashSet; import java.util.Map; import java.util.Set; -import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.SynchronousQueue; @@ -1272,7 +1271,7 @@ public void addToRecvQueue(final 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(final long timeout, final TimeUnit unit) throws InterruptedException {