hbase-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From apurt...@apache.org
Subject [2/4] hbase git commit: HBASE-20017 BufferedMutatorImpl submit the same mutation repeatedly
Date Wed, 21 Feb 2018 01:00:31 GMT
HBASE-20017 BufferedMutatorImpl submit the same mutation repeatedly

Signed-off-by: Andrew Purtell <apurtell@apache.org>


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

Branch: refs/heads/branch-1.4
Commit: 7eaf67a33e36516d8be7a67ce8dadd0d533dc5e4
Parents: f263137
Author: Chia-Ping Tsai <chia7712@gmail.com>
Authored: Sun Feb 18 17:46:08 2018 +0800
Committer: Andrew Purtell <apurtell@apache.org>
Committed: Tue Feb 20 16:59:47 2018 -0800

----------------------------------------------------------------------
 .../hbase/client/BufferedMutatorImpl.java       | 69 +++++++++++---------
 .../hadoop/hbase/client/TestAsyncProcess.java   | 45 +++++++++++++
 .../hadoop/hbase/regionserver/StoreScanner.java |  2 +-
 3 files changed, 83 insertions(+), 33 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hbase/blob/7eaf67a3/hbase-client/src/main/java/org/apache/hadoop/hbase/client/BufferedMutatorImpl.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/BufferedMutatorImpl.java
b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/BufferedMutatorImpl.java
index 1974be3..d207a82 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/BufferedMutatorImpl.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/BufferedMutatorImpl.java
@@ -16,6 +16,7 @@
 package org.apache.hadoop.hbase.client;
 
 import com.google.common.annotations.VisibleForTesting;
+import java.io.Closeable;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
@@ -59,7 +60,7 @@ import java.util.concurrent.atomic.AtomicLong;
 public class BufferedMutatorImpl implements BufferedMutator {
 
   private static final Log LOG = LogFactory.getLog(BufferedMutatorImpl.class);
-  
+
   private final ExceptionListener listener;
 
   protected ClusterConnection connection; // non-final so can be overridden in test
@@ -228,26 +229,22 @@ public class BufferedMutatorImpl implements BufferedMutator {
     }
 
     if (!synchronous) {
-      QueueRowAccess taker = new QueueRowAccess();
-      try {
+      try (QueueRowAccess taker = createQueueRowAccess()){
         ap.submit(tableName, taker, true, null, false);
         if (ap.hasError()) {
           LOG.debug(tableName + ": One or more of the operations have failed -"
               + " waiting for all operation in progress to finish (successfully or not)");
         }
-      } finally {
-        taker.restoreRemainder();
       }
     }
     if (synchronous || ap.hasError()) {
-      QueueRowAccess taker = new QueueRowAccess();
-      try {
-        while (!taker.isEmpty()) {
+      while (true) {
+        try (QueueRowAccess taker = createQueueRowAccess()){
+          if (taker.isEmpty()) {
+            break;
+          }
           ap.submit(tableName, taker, true, null, false);
-          taker.reset();
         }
-      } finally {
-        taker.restoreRemainder();
       }
 
       RetriesExhaustedWithDetailsException error =
@@ -304,36 +301,35 @@ public class BufferedMutatorImpl implements BufferedMutator {
     return Arrays.asList(writeAsyncBuffer.toArray(new Row[0]));
   }
 
-  private class QueueRowAccess implements RowAccess<Row> {
-    private int remainder = undealtMutationCount.getAndSet(0);
+  @VisibleForTesting
+  QueueRowAccess createQueueRowAccess() {
+    return new QueueRowAccess();
+  }
 
-    void reset() {
-      restoreRemainder();
-      remainder = undealtMutationCount.getAndSet(0);
-    }
+  @VisibleForTesting
+  class QueueRowAccess implements RowAccess<Row>, Closeable {
+    private int remainder = undealtMutationCount.getAndSet(0);
+    private Mutation last = null;
 
     @Override
     public Iterator<Row> iterator() {
       return new Iterator<Row>() {
-        private final Iterator<Mutation> iter = writeAsyncBuffer.iterator();
         private int countDown = remainder;
-        private Mutation last = null;
         @Override
         public boolean hasNext() {
-          if (countDown <= 0) {
-            return false;
-          }
-          return iter.hasNext();
+          return countDown > 0;
         }
         @Override
         public Row next() {
+          restoreLastMutation();
           if (!hasNext()) {
             throw new NoSuchElementException();
           }
-          last = iter.next();
+          last = writeAsyncBuffer.poll();
           if (last == null) {
             throw new NoSuchElementException();
           }
+          currentWriteBufferSize.addAndGet(-last.heapSize());
           --countDown;
           return last;
         }
@@ -342,28 +338,37 @@ public class BufferedMutatorImpl implements BufferedMutator {
           if (last == null) {
             throw new IllegalStateException();
           }
-          iter.remove();
-          currentWriteBufferSize.addAndGet(-last.heapSize());
           --remainder;
+          last = null;
         }
       };
     }
 
+    private void restoreLastMutation() {
+      // restore the last mutation since it isn't submitted
+      if (last != null) {
+        writeAsyncBuffer.add(last);
+        currentWriteBufferSize.addAndGet(last.heapSize());
+        last = null;
+      }
+    }
+
     @Override
     public int size() {
       return remainder;
     }
 
-    void restoreRemainder() {
+    @Override
+    public boolean isEmpty() {
+      return remainder <= 0;
+    }
+    @Override
+    public void close() {
+      restoreLastMutation();
       if (remainder > 0) {
         undealtMutationCount.addAndGet(remainder);
         remainder = 0;
       }
     }
-
-    @Override
-    public boolean isEmpty() {
-      return remainder <= 0;
-    }
   }
 }

http://git-wip-us.apache.org/repos/asf/hbase/blob/7eaf67a3/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestAsyncProcess.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestAsyncProcess.java
b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestAsyncProcess.java
index e8c7b73..9ed3001 100644
--- a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestAsyncProcess.java
+++ b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestAsyncProcess.java
@@ -20,6 +20,7 @@
 package org.apache.hadoop.hbase.client;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
@@ -1813,4 +1814,48 @@ public class TestAsyncProcess {
     LOG.debug("Expected to sleep " + expectedSleep + "ms, actually slept " + actualSleep
+ "ms");
     Assert.assertTrue("Slept for too long: " + actualSleep + "ms", actualSleep <= expectedSleep);
   }
+
+  @Test
+  public void testQueueRowAccess() throws Exception {
+    ClusterConnection conn = createHConnection();
+    BufferedMutatorImpl mutator = new BufferedMutatorImpl(conn, null, null,
+      new BufferedMutatorParams(DUMMY_TABLE).writeBufferSize(100000));
+    Put p0 = new Put(DUMMY_BYTES_1).addColumn(DUMMY_BYTES_1, DUMMY_BYTES_1, DUMMY_BYTES_1);
+    Put p1 = new Put(DUMMY_BYTES_2).addColumn(DUMMY_BYTES_2, DUMMY_BYTES_2, DUMMY_BYTES_2);
+    mutator.mutate(p0);
+    BufferedMutatorImpl.QueueRowAccess ra0 = mutator.createQueueRowAccess();
+    // QueueRowAccess should take all undealt mutations
+    assertEquals(0, mutator.undealtMutationCount.get());
+    mutator.mutate(p1);
+    assertEquals(1, mutator.undealtMutationCount.get());
+    BufferedMutatorImpl.QueueRowAccess ra1 = mutator.createQueueRowAccess();
+    // QueueRowAccess should take all undealt mutations
+    assertEquals(0, mutator.undealtMutationCount.get());
+    assertEquals(1, ra0.size());
+    assertEquals(1, ra1.size());
+    Iterator<Row> iter0 = ra0.iterator();
+    Iterator<Row> iter1 = ra1.iterator();
+    assertTrue(iter0.hasNext());
+    assertTrue(iter1.hasNext());
+    // the next() will poll the mutation from inner buffer and update the buffer count
+    assertTrue(iter0.next() == p0);
+    assertEquals(1, mutator.writeAsyncBuffer.size());
+    assertEquals(p1.heapSize(), mutator.currentWriteBufferSize.get());
+    assertTrue(iter1.next() == p1);
+    assertEquals(0, mutator.writeAsyncBuffer.size());
+    assertEquals(0, mutator.currentWriteBufferSize.get());
+    assertFalse(iter0.hasNext());
+    assertFalse(iter1.hasNext());
+    // ra0 doest handle the mutation so the mutation won't be pushed back to buffer
+    iter0.remove();
+    ra0.close();
+    assertEquals(0, mutator.undealtMutationCount.get());
+    assertEquals(0, mutator.writeAsyncBuffer.size());
+    assertEquals(0, mutator.currentWriteBufferSize.get());
+    // ra1 doesn't handle the mutation so the mutation will be pushed back to buffer
+    ra1.close();
+    assertEquals(1, mutator.undealtMutationCount.get());
+    assertEquals(1, mutator.writeAsyncBuffer.size());
+    assertEquals(p1.heapSize(), mutator.currentWriteBufferSize.get());
+  }
 }

http://git-wip-us.apache.org/repos/asf/hbase/blob/7eaf67a3/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
index 632f146..0b2d12f 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
@@ -847,7 +847,7 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
       // these scanners are properly closed() whether or not the scan is completed successfully
       // Eagerly creating scanners so that we have the ref counting ticking on the newly
created
       // store files. In case of stream scanners this eager creation does not induce performance
-      // penalty because in scans (that uses stream scanners) the next() call is bound to
happen.   
+      // penalty because in scans (that uses stream scanners) the next() call is bound to
happen.
       List<KeyValueScanner> scanners = store.getScanners(sfs, cacheBlocks, get, usePread,
         isCompaction, matcher, scan.getStartRow(), scan.getStopRow(), this.readPt, false);
       flushedstoreFileScanners.addAll(scanners);


Mime
View raw message