You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by sy...@apache.org on 2015/12/26 18:07:46 UTC
[13/43] hbase git commit: HBASE-15003 remove
BoundedConcurrentLinkedQueue.
HBASE-15003 remove BoundedConcurrentLinkedQueue.
Signed-off-by: Matteo Bertozzi <ma...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/ba04e037
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/ba04e037
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/ba04e037
Branch: refs/heads/hbase-12439
Commit: ba04e0372da560715bb4b1aa329c0091764340ea
Parents: 58342a8
Author: Sean Busbey <bu...@cloudera.com>
Authored: Thu Dec 17 22:50:19 2015 -0600
Committer: Sean Busbey <bu...@cloudera.com>
Committed: Fri Dec 18 12:26:38 2015 -0600
----------------------------------------------------------------------
.../util/BoundedConcurrentLinkedQueue.java | 122 --------------
.../util/TestBoundedConcurrentLinkedQueue.java | 161 -------------------
2 files changed, 283 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hbase/blob/ba04e037/hbase-server/src/main/java/org/apache/hadoop/hbase/util/BoundedConcurrentLinkedQueue.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/BoundedConcurrentLinkedQueue.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/BoundedConcurrentLinkedQueue.java
deleted file mode 100644
index f66771b..0000000
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/BoundedConcurrentLinkedQueue.java
+++ /dev/null
@@ -1,122 +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.hadoop.hbase.util;
-
-import java.util.Collection;
-import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.atomic.AtomicLong;
-
-import org.apache.hadoop.hbase.classification.InterfaceAudience;
-import org.apache.hadoop.hbase.classification.InterfaceStability;
-
-/**
- * A ConcurrentLinkedQueue that enforces a maximum queue size.
- */
-@InterfaceAudience.Private
-@InterfaceStability.Stable
-public class BoundedConcurrentLinkedQueue<T> extends ConcurrentLinkedQueue<T> {
- private static final long serialVersionUID = 1L;
- private final AtomicLong size = new AtomicLong(0L);
- private final long maxSize;
-
- public BoundedConcurrentLinkedQueue() {
- this(Long.MAX_VALUE);
- }
-
- public BoundedConcurrentLinkedQueue(long maxSize) {
- super();
- this.maxSize = maxSize;
- }
-
- @Override
- public boolean addAll(Collection<? extends T> c) {
- for (;;) {
- long currentSize = size.get();
- long nextSize = currentSize + c.size();
- if (nextSize > maxSize) { // already exceeded limit
- return false;
- }
- if (size.compareAndSet(currentSize, nextSize)) {
- break;
- }
- }
- return super.addAll(c); // Always true for ConcurrentLinkedQueue
- }
-
- @Override
- public void clear() {
- // override this method to batch update size.
- long removed = 0L;
- while (super.poll() != null) {
- removed++;
- }
- size.addAndGet(-removed);
- }
-
- @Override
- public boolean offer(T e) {
- for (;;) {
- long currentSize = size.get();
- if (currentSize >= maxSize) { // already exceeded limit
- return false;
- }
- if (size.compareAndSet(currentSize, currentSize + 1)) {
- break;
- }
- }
- return super.offer(e); // Always true for ConcurrentLinkedQueue
- }
-
- @Override
- public T poll() {
- T result = super.poll();
- if (result != null) {
- size.decrementAndGet();
- }
- return result;
- }
-
- @Override
- public boolean remove(Object o) {
- boolean result = super.remove(o);
- if (result) {
- size.decrementAndGet();
- }
- return result;
- }
-
- @Override
- public int size() {
- return (int) size.get();
- }
-
- public void drainTo(Collection<T> list) {
- long removed = 0;
- for (T element; (element = super.poll()) != null;) {
- list.add(element);
- removed++;
- }
- // Limit the number of operations on size by only reporting size change after the drain is
- // completed.
- size.addAndGet(-removed);
- }
-
- public long remainingCapacity() {
- return maxSize - size.get();
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hbase/blob/ba04e037/hbase-server/src/test/java/org/apache/hadoop/hbase/util/TestBoundedConcurrentLinkedQueue.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/util/TestBoundedConcurrentLinkedQueue.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/util/TestBoundedConcurrentLinkedQueue.java
deleted file mode 100644
index f6e6ac5..0000000
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/util/TestBoundedConcurrentLinkedQueue.java
+++ /dev/null
@@ -1,161 +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.hadoop.hbase.util;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Random;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import org.apache.hadoop.hbase.testclassification.MiscTests;
-import org.apache.hadoop.hbase.testclassification.SmallTests;
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-
-@Category({ MiscTests.class, SmallTests.class })
-public class TestBoundedConcurrentLinkedQueue {
- private final static int CAPACITY = 16;
-
- private BoundedConcurrentLinkedQueue<Long> queue;
-
- @Before
- public void setUp() throws Exception {
- this.queue = new BoundedConcurrentLinkedQueue<Long>(CAPACITY);
- }
-
- @Test
- public void testOfferAndPoll() throws Exception {
- // Offer
- for (long i = 1; i <= CAPACITY; ++i) {
- assertTrue(queue.offer(i));
- assertEquals(i, queue.size());
- assertEquals(CAPACITY - i, queue.remainingCapacity());
- }
- assertFalse(queue.offer(0L));
-
- // Poll
- for (int i = 1; i <= CAPACITY; ++i) {
- long l = queue.poll();
- assertEquals(i, l);
- assertEquals(CAPACITY - i, queue.size());
- assertEquals(i, queue.remainingCapacity());
- }
- assertEquals(null, queue.poll());
- }
-
- @Test
- public void testDrain() throws Exception {
- // Offer
- for (long i = 1; i <= CAPACITY; ++i) {
- assertTrue(queue.offer(i));
- assertEquals(i, queue.size());
- assertEquals(CAPACITY - i, queue.remainingCapacity());
- }
- assertFalse(queue.offer(0L));
-
- // Drain
- List<Long> list = new ArrayList<Long>();
- queue.drainTo(list);
- assertEquals(null, queue.poll());
- assertEquals(0, queue.size());
- assertEquals(CAPACITY, queue.remainingCapacity());
- }
-
- @Test
- public void testClear() {
- // Offer
- for (long i = 1; i <= CAPACITY; ++i) {
- assertTrue(queue.offer(i));
- assertEquals(i, queue.size());
- assertEquals(CAPACITY - i, queue.remainingCapacity());
- }
- assertFalse(queue.offer(0L));
-
- queue.clear();
- assertEquals(null, queue.poll());
- assertEquals(0, queue.size());
- assertEquals(CAPACITY, queue.remainingCapacity());
- }
-
- @Test
- public void testMultiThread() throws InterruptedException {
- int offerThreadCount = 10;
- int pollThreadCount = 5;
- int duration = 5000; // ms
- final AtomicBoolean stop = new AtomicBoolean(false);
- Thread[] offerThreads = new Thread[offerThreadCount];
- for (int i = 0; i < offerThreadCount; i++) {
- offerThreads[i] = new Thread("offer-thread-" + i) {
-
- @Override
- public void run() {
- Random rand = new Random();
- while (!stop.get()) {
- queue.offer(rand.nextLong());
- try {
- Thread.sleep(1);
- } catch (InterruptedException e) {
- }
- }
- }
-
- };
- }
- Thread[] pollThreads = new Thread[pollThreadCount];
- for (int i = 0; i < pollThreadCount; i++) {
- pollThreads[i] = new Thread("poll-thread-" + i) {
-
- @Override
- public void run() {
- while (!stop.get()) {
- queue.poll();
- try {
- Thread.sleep(1);
- } catch (InterruptedException e) {
- }
- }
- }
-
- };
- }
- for (Thread t : offerThreads) {
- t.start();
- }
- for (Thread t : pollThreads) {
- t.start();
- }
- long startTime = System.currentTimeMillis();
- while (System.currentTimeMillis() - startTime < duration) {
- assertTrue(queue.size() <= CAPACITY);
- Thread.yield();
- }
- stop.set(true);
- for (Thread t : offerThreads) {
- t.join();
- }
- for (Thread t : pollThreads) {
- t.join();
- }
- assertTrue(queue.size() <= CAPACITY);
- }
-}