HBASE-2616 TestHRegion.testWritesWhileGetting flaky on trunk

git-svn-id: https://svn.apache.org/repos/asf/hbase/trunk@952796 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -377,6 +377,7 @@ Release 0.21.0 - Unreleased
    HBASE-2614  killing server in TestMasterTransitions causes NPEs and test deadlock
    HBASE-2615  M/R on bulk imported tables
    HBASE-2676  TestInfoServers should use ephemeral ports
+   HBASE-2616  TestHRegion.testWritesWhileGetting flaky on trunk
 
   IMPROVEMENTS
    HBASE-1760  Cleanup TODOs in HTable

src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java

@@ -982,13 +982,6 @@ public class HRegion implements HConstants, HeapSize { // , Writable{
         storeFlushers.add(s.getStoreFlusher(completeSequenceId));
       }
 
-      // This thread is going to cause a whole bunch of scanners to reseek.
-      // They are depending
-      // on a thread-local to know where to read from.
-      // The reason why we set it up high is so that each HRegionScanner only
-      // has a single read point for all its sub-StoreScanners.
-      ReadWriteConsistencyControl.resetThreadReadPoint(rwcc);
-
       // prepare flush (take a snapshot)
       for (StoreFlusher flusher : storeFlushers) {
         flusher.prepare();
@@ -1026,12 +1019,9 @@ public class HRegion implements HConstants, HeapSize { // , Writable{
       }
 
       try {
-        // update this again to make sure we are 'fresh'
-        ReadWriteConsistencyControl.resetThreadReadPoint(rwcc);
-
-        if (atomicWork != null) {
-          atomicWork.call();
-        }
+	if (atomicWork != null) {
+	  atomicWork.call();
+	}
 
         // Switch snapshot (in memstore) -> new hfile (thus causing
         // all the store scanners to reset/reseek).
@@ -1452,7 +1442,6 @@ public class HRegion implements HConstants, HeapSize { // , Writable{
    * @param family
    * @param qualifier
    * @param expectedValue
-   * @param put
    * @param lockId
    * @param writeToWAL
    * @throws IOException
@@ -1949,14 +1938,12 @@ public class HRegion implements HConstants, HeapSize { // , Writable{
     private final byte [] stopRow;
     private Filter filter;
     private List<KeyValue> results = new ArrayList<KeyValue>();
-    private int isScan;
     private int batch;
-    // Doesn't need to be volatile, always accessed under a sync'ed method
+    private int isScan;
     private boolean filterClosed = false;
-    private Scan theScan = null;
-    private List<KeyValueScanner> extraScanners = null;
+    private long readPt;
 
-    RegionScanner(Scan scan, List<KeyValueScanner> additionalScanners) {
+    RegionScanner(Scan scan, List<KeyValueScanner> additionalScanners) throws IOException {
       //DebugPrint.println("HRegionScanner.<init>");
       this.filter = scan.getFilter();
       this.batch = scan.getBatch();
@@ -1968,35 +1955,33 @@ public class HRegion implements HConstants, HeapSize { // , Writable{
       // If we are doing a get, we want to be [startRow,endRow] normally
       // it is [startRow,endRow) and if startRow=endRow we get nothing.
       this.isScan = scan.isGetScan() ? -1 : 0;
-      this.theScan = scan;
-      this.extraScanners = additionalScanners;
-    }
 
-    RegionScanner(Scan scan) {
-      this(scan, null);
-    }
+      this.readPt = ReadWriteConsistencyControl.resetThreadReadPoint(rwcc);
 
-    void initHeap() throws IOException {
       List<KeyValueScanner> scanners = new ArrayList<KeyValueScanner>();
-      if (extraScanners != null) {
-        scanners.addAll(extraScanners);
+      if (additionalScanners != null) {
+        scanners.addAll(additionalScanners);
       }
 
       for (Map.Entry<byte[], NavigableSet<byte[]>> entry :
-          theScan.getFamilyMap().entrySet()) {
+          scan.getFamilyMap().entrySet()) {
         Store store = stores.get(entry.getKey());
-        scanners.add(store.getScanner(theScan, entry.getValue()));
+        scanners.add(store.getScanner(scan, entry.getValue()));
       }
       this.storeHeap = new KeyValueHeap(scanners, comparator);
     }
 
-    private void resetFilters() {
+    RegionScanner(Scan scan) throws IOException {
+      this(scan, null);
+    }
+
+    /**
+     * Reset both the filter and the old filter.
+     */
+    protected void resetFilters() {
       if (filter != null) {
         filter.reset();
       }
-
-      // Start the next row read and reset the thread point
-      ReadWriteConsistencyControl.resetThreadReadPoint(rwcc);
     }
 
     public synchronized boolean next(List<KeyValue> outResults, int limit)
@@ -2013,11 +1998,7 @@ public class HRegion implements HConstants, HeapSize { // , Writable{
       }
 
       // This could be a new thread from the last time we called next().
-      ReadWriteConsistencyControl.resetThreadReadPoint(rwcc);
-      // lazy init the store heap.
-      if (storeHeap == null) {
-        initHeap();
-      }
+      ReadWriteConsistencyControl.setThreadReadPoint(this.readPt);
 
       results.clear();
       boolean returnResult = nextInternal(limit);

src/main/java/org/apache/hadoop/hbase/regionserver/ReadWriteConsistencyControl.java

@@ -22,7 +22,11 @@ public class ReadWriteConsistencyControl {
   public static long getThreadReadPoint() {
     return perThreadReadPoint.get();
   }
-  
+
+  public static void setThreadReadPoint(long readPoint) {
+    perThreadReadPoint.set(readPoint);
+  }
+
   public static long resetThreadReadPoint(ReadWriteConsistencyControl rwcc) {
     perThreadReadPoint.set(rwcc.memstoreReadPoint());
     return getThreadReadPoint();

src/main/java/org/apache/hadoop/hbase/regionserver/Store.java

@@ -1064,11 +1064,8 @@ public class Store implements HConstants, HeapSize {
         if (result != null) {
           newStoreFiles.add(result);
         }
-        this.storefiles = ImmutableList.copyOf(newStoreFiles);
 
-        // WARN ugly hack here, but necessary sadly.
-        // TODO why is this necessary? need a comment here if it's unintuitive!
-        ReadWriteConsistencyControl.resetThreadReadPoint(region.getRWCC());
+	this.storefiles = ImmutableList.copyOf(newStoreFiles);
 
         // Tell observers that list of StoreFiles has changed.
         notifyChangedReadersObservers();

src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java

@@ -42,16 +42,20 @@ class StoreScanner implements KeyValueScanner, InternalScanner, ChangedReadersOb
   private boolean cacheBlocks;
 
   // Used to indicate that the scanner has closed (see HBASE-1107)
+  // Doesnt need to be volatile because it's always accessed via synchronized methods
   private boolean closing = false;
   private final boolean isGet;
 
+  // if heap == null and lastTop != null, you need to reseek given the key below
+  private KeyValue lastTop = null;
+
   /**
    * Opens a scanner across memstore, snapshot, and all StoreFiles.
    *
    * @param store who we scan
    * @param scan the spec
    * @param columns which columns we are scanning
-   * @throws IOException 
+   * @throws IOException
    */
   StoreScanner(Store store, Scan scan, final NavigableSet<byte[]> columns) throws IOException {
     this.store = store;
@@ -125,7 +129,7 @@ class StoreScanner implements KeyValueScanner, InternalScanner, ChangedReadersOb
    */
   private List<KeyValueScanner> getScanners() throws IOException {
     // First the store file scanners
-    
+
     // TODO this used to get the store files in descending order,
     // but now we get them in ascending order, which I think is
     // actually more correct, since memstore get put at the end.
@@ -142,7 +146,7 @@ class StoreScanner implements KeyValueScanner, InternalScanner, ChangedReadersOb
   /*
    * @return List of scanners to seek, possibly filtered by StoreFile.
    */
-  private List<KeyValueScanner> getScanners(Scan scan, 
+  private List<KeyValueScanner> getScanners(Scan scan,
       final NavigableSet<byte[]> columns) throws IOException {
     // First the store file scanners
     List<StoreFileScanner> sfScanners = StoreFileScanner
@@ -152,7 +156,7 @@ class StoreScanner implements KeyValueScanner, InternalScanner, ChangedReadersOb
 
     // exclude scan files that have failed file filters
     for(StoreFileScanner sfs : sfScanners) {
-      if (isGet && 
+      if (isGet &&
           !sfs.getHFileScanner().shouldSeek(scan.getStartRow(), columns)) {
         continue; // exclude this hfs
       }
@@ -165,6 +169,15 @@ class StoreScanner implements KeyValueScanner, InternalScanner, ChangedReadersOb
   }
 
   public synchronized KeyValue peek() {
+    try {
+      checkReseek();
+    } catch (IOException e) {
+      throw new RuntimeException("IOE conversion", e);
+    }
+    if (this.heap == null) {
+      return null;
+    }
+
     return this.heap.peek();
   }
 
@@ -179,10 +192,18 @@ class StoreScanner implements KeyValueScanner, InternalScanner, ChangedReadersOb
     // under test, we dont have a this.store
     if (this.store != null)
       this.store.deleteChangedReaderObserver(this);
-    this.heap.close();
+    if (this.heap != null)
+      this.heap.close();
   }
 
   public synchronized boolean seek(KeyValue key) throws IOException {
+    if (this.heap == null) {
+
+      List<KeyValueScanner> scanners = getScanners();
+
+      heap = new KeyValueHeap(scanners, store.comparator);
+    }
+
     return this.heap.seek(key);
   }
 
@@ -194,11 +215,22 @@ class StoreScanner implements KeyValueScanner, InternalScanner, ChangedReadersOb
    */
   public synchronized boolean next(List<KeyValue> outResult, int limit) throws IOException {
     //DebugPrint.println("SS.next");
+
+    checkReseek();
+
+    // if the heap was left null, then the scanners had previously run out anyways, close and
+    // return.
+    if (this.heap == null) {
+      close();
+      return false;
+    }
+
     KeyValue peeked = this.heap.peek();
     if (peeked == null) {
       close();
       return false;
     }
+
     matcher.setRow(peeked.getRow());
     KeyValue kv;
     List<KeyValue> results = new ArrayList<KeyValue>();
@@ -265,17 +297,37 @@ class StoreScanner implements KeyValueScanner, InternalScanner, ChangedReadersOb
   // Implementation of ChangedReadersObserver
   public synchronized void updateReaders() throws IOException {
     if (this.closing) return;
-    KeyValue topKey = this.peek();
-    if (topKey == null) return;
 
-    List<KeyValueScanner> scanners = getScanners();
+    // this could be null.
+    this.lastTop = this.peek();
 
-    // close the previous scanners:
+    //DebugPrint.println("SS updateReaders, topKey = " + lastTop);
+
+    // close scanners to old obsolete Store files
     this.heap.close(); // bubble thru and close all scanners.
     this.heap = null; // the re-seeks could be slow (access HDFS) free up memory ASAP
 
+    // Let the next() call handle re-creating and seeking
+  }
+
+  private void checkReseek() throws IOException {
+    if (this.heap == null && this.lastTop != null) {
+
+      reseek(this.lastTop);
+      this.lastTop = null; // gone!
+    }
+    // else dont need to reseek
+  }
+
+  private void reseek(KeyValue lastTopKey) throws IOException {
+    if (heap != null) {
+      throw new RuntimeException("StoreScanner.reseek run on an existing heap!");
+    }
+
+    List<KeyValueScanner> scanners = getScanners();
+
     for(KeyValueScanner scanner : scanners) {
-      scanner.seek(topKey);
+      scanner.seek(lastTopKey);
     }
 
     // Combine all seeked scanners with a heap
@@ -284,6 +336,6 @@ class StoreScanner implements KeyValueScanner, InternalScanner, ChangedReadersOb
     // Reset the state of the Query Matcher and set to top row
     matcher.reset();
     KeyValue kv = heap.peek();
-    matcher.setRow((kv == null ? topKey : kv).getRow());
+    matcher.setRow((kv == null ? lastTopKey : kv).getRow());
   }
 }

src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java

@@ -1231,12 +1231,10 @@ public class TestHRegion extends HBaseTestCase {
     scan.addFamily(fam2);
     scan.addFamily(fam4);
     is = (RegionScanner) region.getScanner(scan);
-    is.initHeap(); // i dont like this test
     assertEquals(1, ((RegionScanner)is).storeHeap.getHeap().size());
 
     scan = new Scan();
     is = (RegionScanner) region.getScanner(scan);
-    is.initHeap();
     assertEquals(families.length -1,
         ((RegionScanner)is).storeHeap.getHeap().size());
   }