hbase-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From bus...@apache.org
Subject hbase git commit: HBASE-15003 remove BoundedConcurrentLinkedQueue.
Date Fri, 18 Dec 2015 18:28:25 GMT
Repository: hbase
Updated Branches:
  refs/heads/branch-1 299ade435 -> d1c412ae8


HBASE-15003 remove BoundedConcurrentLinkedQueue.

Signed-off-by: Matteo Bertozzi <matteo.bertozzi@cloudera.com>

Conflicts:
	hbase-server/src/test/java/org/apache/hadoop/hbase/util/TestBoundedConcurrentLinkedQueue.java


Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/d1c412ae
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/d1c412ae
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/d1c412ae

Branch: refs/heads/branch-1
Commit: d1c412ae8f714c56eede8f0aca8d9c37bc1ed97a
Parents: 299ade4
Author: Sean Busbey <busbey@cloudera.com>
Authored: Thu Dec 17 22:50:19 2015 -0600
Committer: Sean Busbey <busbey@cloudera.com>
Committed: Fri Dec 18 12:28:02 2015 -0600

----------------------------------------------------------------------
 .../util/BoundedConcurrentLinkedQueue.java      | 122 --------------
 .../util/TestBoundedConcurrentLinkedQueue.java  | 160 -------------------
 2 files changed, 282 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hbase/blob/d1c412ae/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/d1c412ae/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 d3c206f..0000000
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/util/TestBoundedConcurrentLinkedQueue.java
+++ /dev/null
@@ -1,160 +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.SmallTests;
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-
-@Category(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);
-  }
-}


Mime
View raw message