hbase-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mberto...@apache.org
Subject hbase git commit: HBASE-16642 Use DelayQueue instead of TimeoutBlockingQueue
Date Fri, 14 Oct 2016 04:55:01 GMT
Repository: hbase
Updated Branches:
  refs/heads/master 91a7bbd58 -> 9a94dc90b


HBASE-16642 Use DelayQueue instead of TimeoutBlockingQueue

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


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

Branch: refs/heads/master
Commit: 9a94dc90b4a53ca9d82c662f19c0369d3cd9aecf
Parents: 91a7bbd
Author: Hiroshi Ikeda <ikeda@vic.co.jp>
Authored: Thu Oct 13 20:21:46 2016 -0700
Committer: Matteo Bertozzi <matteo.bertozzi@cloudera.com>
Committed: Thu Oct 13 21:41:54 2016 -0700

----------------------------------------------------------------------
 .../hbase/procedure2/ProcedureExecutor.java     |  93 +++++---
 .../procedure2/util/TimeoutBlockingQueue.java   | 234 -------------------
 .../util/TestTimeoutBlockingQueue.java          | 159 -------------
 3 files changed, 67 insertions(+), 419 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hbase/blob/9a94dc90/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/ProcedureExecutor.java
----------------------------------------------------------------------
diff --git a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/ProcedureExecutor.java
b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/ProcedureExecutor.java
index 2e9e3a3..14fe71b 100644
--- a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/ProcedureExecutor.java
+++ b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/ProcedureExecutor.java
@@ -29,12 +29,15 @@ import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.DelayQueue;
+import java.util.concurrent.Delayed;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.logging.Log;
@@ -48,8 +51,6 @@ import org.apache.hadoop.hbase.exceptions.IllegalArgumentIOException;
 import org.apache.hadoop.hbase.procedure2.store.ProcedureStore;
 import org.apache.hadoop.hbase.procedure2.store.ProcedureStore.ProcedureIterator;
 import org.apache.hadoop.hbase.procedure2.util.StringUtils;
-import org.apache.hadoop.hbase.procedure2.util.TimeoutBlockingQueue;
-import org.apache.hadoop.hbase.procedure2.util.TimeoutBlockingQueue.TimeoutRetriever;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.ProcedureProtos.ProcedureState;
 import org.apache.hadoop.hbase.security.User;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
@@ -96,17 +97,58 @@ public class ProcedureExecutor<TEnvironment> {
   }
 
   /**
-   * Used by the TimeoutBlockingQueue to get the timeout interval of the procedure
+   * Used by the DelayQueue to get the timeout interval of the procedure
    */
-  private static class ProcedureTimeoutRetriever implements TimeoutRetriever<Procedure>
{
+  private static class DelayedContainer implements Delayed {
+    static final DelayedContainer POISON = new DelayedContainer();
+
+    /** null if poison */
+    final Procedure proc;
+    final long timeoutTime;
+
+    DelayedContainer(Procedure proc) {
+      assert proc != null;
+      this.proc = proc;
+      this.timeoutTime = proc.getLastUpdate() + proc.getTimeout();
+    }
+
+    DelayedContainer() {
+      this.proc = null;
+      this.timeoutTime = Long.MIN_VALUE;
+    }
+
+    @Override
+    public long getDelay(TimeUnit unit) {
+      long currentTime = EnvironmentEdgeManager.currentTime();
+      if (currentTime >= timeoutTime) {
+        return 0;
+      }
+      return unit.convert(timeoutTime - currentTime, TimeUnit.MICROSECONDS);
+    }
+
+    /**
+     * @throws NullPointerException {@inheritDoc}
+     * @throws ClassCastException {@inheritDoc}
+     */
+    @Override
+    public int compareTo(Delayed o) {
+      return Long.compare(timeoutTime, ((DelayedContainer)o).timeoutTime);
+    }
+
     @Override
-    public long getTimeout(Procedure proc) {
-      return proc.getTimeRemaining();
+    public boolean equals(Object obj) {
+      if (obj == this) {
+        return true;
+      }
+      if (! (obj instanceof DelayedContainer)) {
+        return false;
+      }
+      return Objects.equals(proc, ((DelayedContainer)obj).proc);
     }
 
     @Override
-    public TimeUnit getTimeUnit(Procedure proc) {
-      return TimeUnit.MILLISECONDS;
+    public int hashCode() {
+      return proc != null ? proc.hashCode() : 0;
     }
   }
 
@@ -239,8 +281,8 @@ public class ProcedureExecutor<TEnvironment> {
    * Timeout Queue that contains Procedures in a WAITING_TIMEOUT state
    * or periodic procedures.
    */
-  private final TimeoutBlockingQueue<Procedure> waitingTimeout =
-    new TimeoutBlockingQueue<Procedure>(new ProcedureTimeoutRetriever());
+  private final DelayQueue<DelayedContainer> waitingTimeout =
+    new DelayQueue<DelayedContainer>();
 
   /**
    * Scheduler/Queue that contains runnable procedures.
@@ -544,7 +586,7 @@ public class ProcedureExecutor<TEnvironment> {
 
     LOG.info("Stopping the procedure executor");
     scheduler.stop();
-    waitingTimeout.signalAll();
+    waitingTimeout.add(DelayedContainer.POISON);
   }
 
   public void join() {
@@ -628,7 +670,7 @@ public class ProcedureExecutor<TEnvironment> {
    */
   public void addChore(final ProcedureInMemoryChore chore) {
     chore.setState(ProcedureState.RUNNABLE);
-    waitingTimeout.add(chore);
+    waitingTimeout.add(new DelayedContainer(chore));
   }
 
   /**
@@ -638,7 +680,7 @@ public class ProcedureExecutor<TEnvironment> {
    */
   public boolean removeChore(final ProcedureInMemoryChore chore) {
     chore.setState(ProcedureState.FINISHED);
-    return waitingTimeout.remove(chore);
+    return waitingTimeout.remove(new DelayedContainer(chore));
   }
 
   /**
@@ -927,15 +969,16 @@ public class ProcedureExecutor<TEnvironment> {
 
   private void timeoutLoop() {
     while (isRunning()) {
-      Procedure proc = waitingTimeout.poll();
-      if (proc == null) continue;
-
-      if (proc.getTimeRemaining() > 100) {
-        // got an early wake, maybe a stop?
-        // re-enqueue the task in case was not a stop or just a signal
-        waitingTimeout.add(proc);
+      Procedure proc;
+      try {
+        proc = waitingTimeout.take().proc;
+      } catch (InterruptedException e) {
+        // Just consume the interruption.
         continue;
       }
+      if (proc == null) { // POISON to stop
+        break;
+      }
 
       // ----------------------------------------------------------------------------
       // TODO-MAYBE: Should we provide a notification to the store with the
@@ -955,8 +998,8 @@ public class ProcedureExecutor<TEnvironment> {
           } catch (Throwable e) {
             LOG.error("Ignoring CompletedProcedureCleaner exception: " + e.getMessage(),
e);
           }
-          proc.setStartTime(EnvironmentEdgeManager.currentTime());
-          if (proc.isRunnable()) waitingTimeout.add(proc);
+          proc.updateTimestamp();
+          if (proc.isRunnable()) waitingTimeout.add(new DelayedContainer(proc));
         }
         continue;
       }
@@ -970,8 +1013,6 @@ public class ProcedureExecutor<TEnvironment> {
         store.update(proc);
         scheduler.addFront(proc);
         continue;
-      } else if (proc.getState() == ProcedureState.WAITING_TIMEOUT) {
-        waitingTimeout.add(proc);
       }
     }
   }
@@ -1171,7 +1212,7 @@ public class ProcedureExecutor<TEnvironment> {
                   procedure.setState(ProcedureState.WAITING);
                   break;
                 case WAITING_TIMEOUT:
-                  waitingTimeout.add(procedure);
+                  waitingTimeout.add(new DelayedContainer(procedure));
                   break;
                 default:
                   break;
@@ -1179,7 +1220,7 @@ public class ProcedureExecutor<TEnvironment> {
             }
           }
         } else if (procedure.getState() == ProcedureState.WAITING_TIMEOUT) {
-          waitingTimeout.add(procedure);
+          waitingTimeout.add(new DelayedContainer(procedure));
         } else if (!isSuspended) {
           // No subtask, so we are done
           procedure.setState(ProcedureState.FINISHED);

http://git-wip-us.apache.org/repos/asf/hbase/blob/9a94dc90/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/util/TimeoutBlockingQueue.java
----------------------------------------------------------------------
diff --git a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/util/TimeoutBlockingQueue.java
b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/util/TimeoutBlockingQueue.java
deleted file mode 100644
index 2292e63..0000000
--- a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/util/TimeoutBlockingQueue.java
+++ /dev/null
@@ -1,234 +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.procedure2.util;
-
-import java.util.concurrent.locks.Condition;
-import java.util.concurrent.locks.ReentrantLock;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.hadoop.hbase.classification.InterfaceAudience;
-import org.apache.hadoop.hbase.classification.InterfaceStability;
-
-@InterfaceAudience.Private
-@InterfaceStability.Evolving
-public class TimeoutBlockingQueue<E> {
-  public static interface TimeoutRetriever<T> {
-    long getTimeout(T object);
-    TimeUnit getTimeUnit(T object);
-  }
-
-  private final ReentrantLock lock = new ReentrantLock();
-  private final Condition waitCond = lock.newCondition();
-  private final TimeoutRetriever<? super E> timeoutRetriever;
-
-  private E[] objects;
-  private int head = 0;
-  private int tail = 0;
-
-  public TimeoutBlockingQueue(TimeoutRetriever<? super E> timeoutRetriever) {
-    this(32, timeoutRetriever);
-  }
-
-  @SuppressWarnings("unchecked")
-  public TimeoutBlockingQueue(int capacity, TimeoutRetriever<? super E> timeoutRetriever)
{
-    this.objects = (E[])new Object[capacity];
-    this.timeoutRetriever = timeoutRetriever;
-  }
-
-  public void dump() {
-    for (int i = 0; i < objects.length; ++i) {
-      if (i == head) {
-        System.out.print("[" + objects[i] + "] ");
-      } else if (i == tail) {
-        System.out.print("]" + objects[i] + "[ ");
-      } else {
-        System.out.print(objects[i] + " ");
-      }
-    }
-    System.out.println();
-  }
-
-  public void clear() {
-    lock.lock();
-    try {
-      if (head != tail) {
-        for (int i = head; i < tail; ++i) {
-          objects[i] = null;
-        }
-        head = 0;
-        tail = 0;
-        waitCond.signal();
-      }
-    } finally {
-      lock.unlock();
-    }
-  }
-
-  public void add(E e) {
-    if (e == null) throw new NullPointerException();
-
-    lock.lock();
-    try {
-      addElement(e);
-      waitCond.signal();
-    } finally {
-      lock.unlock();
-    }
-  }
-
-  public boolean remove(E e) {
-    if (e == null) return false;
-    lock.lock();
-    try {
-      for (int i = 0; i < objects.length; ++i) {
-        if (e.equals(objects[i])) {
-          objects[i] = null;
-          return true;
-        }
-      }
-      return false;
-    } finally {
-      lock.unlock();
-    }
-  }
-
-  @edu.umd.cs.findbugs.annotations.SuppressWarnings("WA_AWAIT_NOT_IN_LOOP")
-  public E poll() {
-    lock.lock();
-    try {
-      if (isEmpty()) {
-        waitCond.await();
-        return null;
-      }
-
-      E elem = objects[head];
-      long nanos = getNanosTimeout(elem);
-      nanos = waitCond.awaitNanos(nanos);
-      return nanos > 0 ? null : removeFirst();
-    } catch (InterruptedException e) {
-      Thread.currentThread().interrupt();
-      return null;
-    } finally {
-      lock.unlock();
-    }
-  }
-
-  public int size() {
-    return tail - head;
-  }
-
-  public boolean isEmpty() {
-    return (tail - head) == 0;
-  }
-
-  public void signalAll() {
-    lock.lock();
-    try {
-      waitCond.signalAll();
-    } finally {
-      lock.unlock();
-    }
-  }
-
-  private void addElement(E elem) {
-    int size = (tail - head);
-    if ((objects.length - size) == 0) {
-      int capacity = size + ((size < 64) ? (size + 2) : (size >> 1));
-      E[] newObjects = (E[])new Object[capacity];
-
-      if (compareTimeouts(objects[tail - 1], elem) <= 0) {
-        // Append
-        System.arraycopy(objects, head, newObjects, 0, tail);
-        tail -= head;
-        newObjects[tail++] = elem;
-      } else if (compareTimeouts(objects[head], elem) > 0) {
-        // Prepend
-        System.arraycopy(objects, head, newObjects, 1, tail);
-        newObjects[0] = elem;
-        tail -= (head - 1);
-      } else {
-        // Insert in the middle
-        int index = upperBound(head, tail - 1, elem);
-        int newIndex = (index - head);
-        System.arraycopy(objects, head, newObjects, 0, newIndex);
-        newObjects[newIndex] = elem;
-        System.arraycopy(objects, index, newObjects, newIndex + 1, tail - index);
-        tail -= (head - 1);
-      }
-      head = 0;
-      objects = newObjects;
-    } else {
-      if (tail == objects.length) {
-        // shift down |-----AAAAAAA|
-        tail -= head;
-        System.arraycopy(objects, head, objects, 0, tail);
-        head = 0;
-      }
-
-      if (tail == head || compareTimeouts(objects[tail - 1], elem) <= 0) {
-        // Append
-        objects[tail++] = elem;
-      } else if (head > 0 && compareTimeouts(objects[head], elem) > 0) {
-        // Prepend
-        objects[--head] = elem;
-      } else {
-        // Insert in the middle
-        int index = upperBound(head, tail - 1, elem);
-        System.arraycopy(objects, index, objects, index + 1, tail - index);
-        objects[index] = elem;
-        tail++;
-      }
-    }
-  }
-
-  private E removeFirst() {
-    E elem = objects[head];
-    objects[head] = null;
-    head = (head + 1) % objects.length;
-    if (head == 0) tail = 0;
-    return elem;
-  }
-
-  private int upperBound(int start, int end, E key) {
-    while (start < end) {
-      int mid = (start + end) >>> 1;
-      E mitem = objects[mid];
-      int cmp = compareTimeouts(mitem, key);
-      if (cmp > 0) {
-        end = mid;
-      } else {
-        start = mid + 1;
-      }
-    }
-    return start;
-  }
-
-  private int compareTimeouts(final E a, final E b) {
-    long t1 = getNanosTimeout(a);
-    long t2 = getNanosTimeout(b);
-    return (t1 < t2) ? -1 : (t1 > t2) ? 1 : 0;
-  }
-
-  private long getNanosTimeout(final E obj) {
-    if (obj == null) return 0;
-    TimeUnit unit = timeoutRetriever.getTimeUnit(obj);
-    long timeout = timeoutRetriever.getTimeout(obj);
-    return unit.toNanos(timeout);
-  }
-}

http://git-wip-us.apache.org/repos/asf/hbase/blob/9a94dc90/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/util/TestTimeoutBlockingQueue.java
----------------------------------------------------------------------
diff --git a/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/util/TestTimeoutBlockingQueue.java
b/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/util/TestTimeoutBlockingQueue.java
deleted file mode 100644
index 1f901b5..0000000
--- a/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/util/TestTimeoutBlockingQueue.java
+++ /dev/null
@@ -1,159 +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.procedure2.util;
-
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-
-import java.util.Arrays;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.hadoop.hbase.CategoryBasedTimeout;
-import org.apache.hadoop.hbase.procedure2.util.TimeoutBlockingQueue.TimeoutRetriever;
-import org.apache.hadoop.hbase.testclassification.MasterTests;
-import org.apache.hadoop.hbase.testclassification.MediumTests;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-import org.junit.rules.TestRule;
-
-@Category({MasterTests.class, MediumTests.class})
-public class TestTimeoutBlockingQueue {
-  @Rule public final TestRule timeout = CategoryBasedTimeout.builder().withTimeout(this.getClass()).
-      withLookingForStuckThread(true).build();
-  static class TestObject {
-    private long timeout;
-    private int seqId;
-
-    public TestObject(int seqId, long timeout) {
-      this.timeout = timeout;
-      this.seqId = seqId;
-    }
-
-    public long getTimeout() {
-      return timeout;
-    }
-
-    public String toString() {
-      return String.format("(%03d, %03d)", seqId, timeout);
-    }
-  }
-
-  static class TestObjectTimeoutRetriever implements TimeoutRetriever<TestObject> {
-    @Override
-    public long getTimeout(TestObject obj) {
-      return obj.getTimeout();
-    }
-
-    @Override
-    public TimeUnit getTimeUnit(TestObject obj) {
-      return TimeUnit.MILLISECONDS;
-    }
-  }
-
-  @Test
-  public void testOrder() {
-    TimeoutBlockingQueue<TestObject> queue =
-      new TimeoutBlockingQueue<TestObject>(8, new TestObjectTimeoutRetriever());
-
-    long[] timeouts = new long[] {500, 200, 700, 300, 600, 600, 200, 800, 500};
-
-    for (int i = 0; i < timeouts.length; ++i) {
-      for (int j = 0; j <= i; ++j) {
-        queue.add(new TestObject(j, timeouts[j]));
-        queue.dump();
-      }
-
-      long prev = 0;
-      for (int j = 0; j <= i; ++j) {
-        TestObject obj = queue.poll();
-        assertTrue(obj.getTimeout() >= prev);
-        prev = obj.getTimeout();
-        queue.dump();
-      }
-    }
-  }
-
-  @Test
-  public void testTimeoutBlockingQueue() {
-    TimeoutBlockingQueue<TestObject> queue;
-
-    int[][] testArray = new int[][] {
-      {200, 400, 600},  // append
-      {200, 400, 100},  // prepend
-      {200, 400, 300},  // insert
-    };
-
-    for (int i = 0; i < testArray.length; ++i) {
-      int[] sortedArray = Arrays.copyOf(testArray[i], testArray[i].length);
-      Arrays.sort(sortedArray);
-
-      // test with head == 0
-      queue = new TimeoutBlockingQueue<TestObject>(2, new TestObjectTimeoutRetriever());
-      for (int j = 0; j < testArray[i].length; ++j) {
-        queue.add(new TestObject(j, testArray[i][j]));
-        queue.dump();
-      }
-
-      for (int j = 0; !queue.isEmpty(); ++j) {
-        assertEquals(sortedArray[j], queue.poll().getTimeout());
-      }
-
-      queue = new TimeoutBlockingQueue<TestObject>(2, new TestObjectTimeoutRetriever());
-      queue.add(new TestObject(0, 50));
-      assertEquals(50, queue.poll().getTimeout());
-
-      // test with head > 0
-      for (int j = 0; j < testArray[i].length; ++j) {
-        queue.add(new TestObject(j, testArray[i][j]));
-        queue.dump();
-      }
-
-      for (int j = 0; !queue.isEmpty(); ++j) {
-        assertEquals(sortedArray[j], queue.poll().getTimeout());
-      }
-    }
-  }
-
-  @Test
-  public void testRemove() {
-    TimeoutBlockingQueue<TestObject> queue =
-      new TimeoutBlockingQueue<TestObject>(2, new TestObjectTimeoutRetriever());
-
-    final int effectiveLen = 5;
-    TestObject[] objs = new TestObject[6];
-    for (int i = 0; i < effectiveLen; ++i) {
-      objs[i] = new TestObject(0, i * 10);
-      queue.add(objs[i]);
-    }
-    objs[effectiveLen] = new TestObject(0, effectiveLen * 10);
-    queue.dump();
-
-    for (int i = 0; i < effectiveLen; i += 2) {
-      assertTrue(queue.remove(objs[i]));
-    }
-    assertTrue(!queue.remove(objs[effectiveLen]));
-
-    for (int i = 0; i < effectiveLen; ++i) {
-      TestObject x = queue.poll();
-      assertEquals((i % 2) == 0 ? null : objs[i], x);
-    }
-  }
-}


Mime
View raw message