View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  
20  package org.apache.hadoop.hbase.regionserver;
21  
22  import java.lang.management.ManagementFactory;
23  import java.lang.management.RuntimeMXBean;
24  import java.util.ArrayList;
25  import java.util.Collections;
26  import java.util.Iterator;
27  import java.util.List;
28  import java.util.NavigableSet;
29  import java.util.SortedSet;
30  import java.util.concurrent.atomic.AtomicInteger;
31  import java.util.concurrent.atomic.AtomicLong;
32  
33  import org.apache.commons.logging.Log;
34  import org.apache.commons.logging.LogFactory;
35  import org.apache.hadoop.hbase.classification.InterfaceAudience;
36  import org.apache.hadoop.conf.Configuration;
37  import org.apache.hadoop.hbase.Cell;
38  import org.apache.hadoop.hbase.CellUtil;
39  import org.apache.hadoop.hbase.HBaseConfiguration;
40  import org.apache.hadoop.hbase.HConstants;
41  import org.apache.hadoop.hbase.KeyValue;
42  import org.apache.hadoop.hbase.KeyValueUtil;
43  import org.apache.hadoop.hbase.client.Scan;
44  import org.apache.hadoop.hbase.io.TimeRange;
45  import org.apache.hadoop.hbase.util.ByteRange;
46  import org.apache.hadoop.hbase.util.Bytes;
47  import org.apache.hadoop.hbase.util.ClassSize;
48  import org.apache.hadoop.hbase.util.CollectionBackedScanner;
49  import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
50  import org.apache.hadoop.hbase.util.ReflectionUtils;
51  import org.apache.htrace.Trace;
52  
53  import com.google.common.annotations.VisibleForTesting;
54  
55  /**
56   * The MemStore holds in-memory modifications to the Store.  Modifications
57   * are {@link Cell}s.  When asked to flush, current memstore is moved
58   * to snapshot and is cleared.  We continue to serve edits out of new memstore
59   * and backing snapshot until flusher reports in that the flush succeeded. At
60   * this point we let the snapshot go.
61   *  <p>
62   * The MemStore functions should not be called in parallel. Callers should hold
63   *  write and read locks. This is done in {@link HStore}.
64   *  </p>
65   *
66   * TODO: Adjust size of the memstore when we remove items because they have
67   * been deleted.
68   * TODO: With new KVSLS, need to make sure we update HeapSize with difference
69   * in KV size.
70   */
71  @InterfaceAudience.Private
72  public class DefaultMemStore implements MemStore {
73    private static final Log LOG = LogFactory.getLog(DefaultMemStore.class);
74    @VisibleForTesting
75    static final String USEMSLAB_KEY = "hbase.hregion.memstore.mslab.enabled";
76    private static final boolean USEMSLAB_DEFAULT = true;
77    private static final String MSLAB_CLASS_NAME = "hbase.regionserver.mslab.class";
78  
79    private Configuration conf;
80  
81    @VisibleForTesting
82    final KeyValue.KVComparator comparator;
83  
84    // Used to track when to flush
85    private volatile long timeOfOldestEdit = Long.MAX_VALUE;
86  
87    private volatile long snapshotId;
88    private volatile boolean tagsPresent;
89  
90    @VisibleForTesting
91    volatile Section activeSection;
92    @VisibleForTesting
93    volatile Section snapshotSection;
94  
95    /**
96     * Default constructor. Used for tests.
97     */
98    public DefaultMemStore() {
99      this(HBaseConfiguration.create(), KeyValue.COMPARATOR);
100   }
101 
102   /**
103    * Constructor.
104    * @param c Comparator
105    */
106   public DefaultMemStore(final Configuration conf,
107                   final KeyValue.KVComparator c) {
108     this.conf = conf;
109     this.comparator = c;
110     this.activeSection = Section.newActiveSection(comparator, conf);
111     this.snapshotSection = Section.newSnapshotSection(comparator);
112   }
113 
114   /**
115    * Creates a snapshot of the current memstore.
116    * Snapshot must be cleared by call to {@link #clearSnapshot(long)}
117    */
118   @Override
119   public MemStoreSnapshot snapshot() {
120     // If snapshot currently has entries, then flusher failed or didn't call
121     // cleanup.  Log a warning.
122     if (!snapshotSection.getCellSkipListSet().isEmpty()) {
123       LOG.warn("Snapshot called again without clearing previous. " +
124           "Doing nothing. Another ongoing flush or did we fail last attempt?");
125     } else {
126       this.snapshotId = EnvironmentEdgeManager.currentTime();
127       if (!activeSection.getCellSkipListSet().isEmpty()) {
128         snapshotSection = activeSection;
129         activeSection = Section.newActiveSection(comparator, conf);
130         snapshotSection.getHeapSize().addAndGet(-DEEP_OVERHEAD);
131         timeOfOldestEdit = Long.MAX_VALUE;
132       }
133     }
134     MemStoreSnapshot memStoreSnapshot = new MemStoreSnapshot(this.snapshotId,
135         snapshotSection.getCellsCount().get(), snapshotSection.getHeapSize().get(),
136         snapshotSection.getTimeRangeTracker(),
137         new CollectionBackedScanner(snapshotSection.getCellSkipListSet(), this.comparator),
138         this.tagsPresent);
139     this.tagsPresent = false;
140     return memStoreSnapshot;
141   }
142 
143   /**
144    * The passed snapshot was successfully persisted; it can be let go.
145    * @param id Id of the snapshot to clean out.
146    * @throws UnexpectedStateException
147    * @see #snapshot()
148    */
149   @Override
150   public void clearSnapshot(long id) throws UnexpectedStateException {
151     if (this.snapshotId == -1) return;  // already cleared
152     if (this.snapshotId != id) {
153       throw new UnexpectedStateException("Current snapshot id is " + this.snapshotId + ",passed "
154           + id);
155     }
156     // OK. Passed in snapshot is same as current snapshot.
157     MemStoreLAB tmpAllocator = snapshotSection.getMemStoreLAB();
158     snapshotSection = Section.newSnapshotSection(comparator);
159     if (tmpAllocator != null) {
160       tmpAllocator.close();
161     }
162     this.snapshotId = -1;
163   }
164 
165   @Override
166   public long getFlushableSize() {
167     long snapshotSize = snapshotSection.getHeapSize().get();
168     return snapshotSize > 0 ? snapshotSize : keySize();
169   }
170 
171   @Override
172   public long getSnapshotSize() {
173     return snapshotSection.getHeapSize().get();
174   }
175 
176   /**
177    * Write an update
178    * @param cell
179    * @return approximate size of the passed cell.
180    */
181   @Override
182   public long add(Cell cell) {
183     Cell toAdd = maybeCloneWithAllocator(cell);
184     boolean mslabUsed = (toAdd != cell);
185     return internalAdd(toAdd, mslabUsed);
186   }
187 
188   @Override
189   public long timeOfOldestEdit() {
190     return timeOfOldestEdit;
191   }
192 
193   private boolean addToCellSet(Cell e) {
194     boolean b = this.activeSection.getCellSkipListSet().add(e);
195     // In no tags case this NoTagsKeyValue.getTagsLength() is a cheap call.
196     // When we use ACL CP or Visibility CP which deals with Tags during
197     // mutation, the TagRewriteCell.getTagsLength() is a cheaper call. We do not
198     // parse the byte[] to identify the tags length.
199     if(e.getTagsLength() > 0) {
200       tagsPresent = true;
201     }
202     setOldestEditTimeToNow();
203     return b;
204   }
205 
206   private boolean removeFromCellSet(Cell e) {
207     boolean b = this.activeSection.getCellSkipListSet().remove(e);
208     setOldestEditTimeToNow();
209     return b;
210   }
211 
212   void setOldestEditTimeToNow() {
213     if (timeOfOldestEdit == Long.MAX_VALUE) {
214       timeOfOldestEdit = EnvironmentEdgeManager.currentTime();
215     }
216   }
217 
218   /**
219    * Internal version of add() that doesn't clone Cells with the
220    * allocator, and doesn't take the lock.
221    *
222    * Callers should ensure they already have the read lock taken
223    * @param toAdd the cell to add
224    * @param mslabUsed whether using MSLAB
225    * @return the heap size change in bytes
226    */
227   private long internalAdd(final Cell toAdd, boolean mslabUsed) {
228     boolean notPresent = addToCellSet(toAdd);
229     long s = heapSizeChange(toAdd, notPresent);
230     if (notPresent) {
231       activeSection.getCellsCount().incrementAndGet();
232     }
233     // If there's already a same cell in the CellSet and we are using MSLAB, we must count in the
234     // MSLAB allocation size as well, or else there will be memory leak (occupied heap size larger
235     // than the counted number)
236     if (!notPresent && mslabUsed) {
237       s += getCellLength(toAdd);
238     }
239     activeSection.getTimeRangeTracker().includeTimestamp(toAdd);
240     activeSection.getHeapSize().addAndGet(s);
241     return s;
242   }
243 
244   /**
245    * Get cell length after serialized in {@link KeyValue}
246    */
247   @VisibleForTesting
248   int getCellLength(Cell cell) {
249     return KeyValueUtil.length(cell);
250   }
251 
252   private Cell maybeCloneWithAllocator(Cell cell) {
253     if (activeSection.getMemStoreLAB() == null) {
254       return cell;
255     }
256 
257     int len = getCellLength(cell);
258     ByteRange alloc = activeSection.getMemStoreLAB().allocateBytes(len);
259     if (alloc == null) {
260       // The allocation was too large, allocator decided
261       // not to do anything with it.
262       return cell;
263     }
264     assert alloc.getBytes() != null;
265     KeyValueUtil.appendToByteArray(cell, alloc.getBytes(), alloc.getOffset());
266     KeyValue newKv = new KeyValue(alloc.getBytes(), alloc.getOffset(), len);
267     newKv.setSequenceId(cell.getSequenceId());
268     return newKv;
269   }
270 
271   /**
272    * Remove n key from the memstore. Only cells that have the same key and the
273    * same memstoreTS are removed.  It is ok to not update timeRangeTracker
274    * in this call. It is possible that we can optimize this method by using
275    * tailMap/iterator, but since this method is called rarely (only for
276    * error recovery), we can leave those optimization for the future.
277    * @param cell
278    */
279   @Override
280   public void rollback(Cell cell) {
281     // If the key is in the snapshot, delete it. We should not update
282     // this.size, because that tracks the size of only the memstore and
283     // not the snapshot. The flush of this snapshot to disk has not
284     // yet started because Store.flush() waits for all rwcc transactions to
285     // commit before starting the flush to disk.
286     Cell found = snapshotSection.getCellSkipListSet().get(cell);
287     if (found != null && found.getSequenceId() == cell.getSequenceId()) {
288       snapshotSection.getCellSkipListSet().remove(cell);
289       long sz = heapSizeChange(cell, true);
290       snapshotSection.getHeapSize().addAndGet(-sz);
291       snapshotSection.getCellsCount().decrementAndGet();
292     }
293     // If the key is in the memstore, delete it. Update this.size.
294     found = activeSection.getCellSkipListSet().get(cell);
295     if (found != null && found.getSequenceId() == cell.getSequenceId()) {
296       removeFromCellSet(cell);
297       long sz = heapSizeChange(found, true);
298       activeSection.getHeapSize().addAndGet(-sz);
299       activeSection.getCellsCount().decrementAndGet();
300     }
301   }
302 
303   /**
304    * Write a delete
305    * @param deleteCell
306    * @return approximate size of the passed key and value.
307    */
308   @Override
309   public long delete(Cell deleteCell) {
310     Cell toAdd = maybeCloneWithAllocator(deleteCell);
311     boolean mslabUsed = (toAdd != deleteCell);
312     return internalAdd(toAdd, mslabUsed);
313   }
314 
315   /**
316    * @param cell Find the row that comes after this one.  If null, we return the
317    * first.
318    * @return Next row or null if none found.
319    */
320   Cell getNextRow(final Cell cell) {
321     return getLowest(getNextRow(cell, activeSection.getCellSkipListSet()),
322           getNextRow(cell, snapshotSection.getCellSkipListSet()));
323   }
324 
325   /*
326    * @param a
327    * @param b
328    * @return Return lowest of a or b or null if both a and b are null
329    */
330   private Cell getLowest(final Cell a, final Cell b) {
331     if (a == null) {
332       return b;
333     }
334     if (b == null) {
335       return a;
336     }
337     return comparator.compareRows(a, b) <= 0? a: b;
338   }
339 
340   /*
341    * @param key Find row that follows this one.  If null, return first.
342    * @param map Set to look in for a row beyond <code>row</code>.
343    * @return Next row or null if none found.  If one found, will be a new
344    * KeyValue -- can be destroyed by subsequent calls to this method.
345    */
346   private Cell getNextRow(final Cell key,
347       final NavigableSet<Cell> set) {
348     Cell result = null;
349     SortedSet<Cell> tail = key == null? set: set.tailSet(key);
350     // Iterate until we fall into the next row; i.e. move off current row
351     for (Cell cell: tail) {
352       if (comparator.compareRows(cell, key) <= 0)
353         continue;
354       // Note: Not suppressing deletes or expired cells.  Needs to be handled
355       // by higher up functions.
356       result = cell;
357       break;
358     }
359     return result;
360   }
361 
362   /**
363    * @param state column/delete tracking state
364    */
365   @Override
366   public void getRowKeyAtOrBefore(final GetClosestRowBeforeTracker state) {
367     getRowKeyAtOrBefore(activeSection.getCellSkipListSet(), state);
368     getRowKeyAtOrBefore(snapshotSection.getCellSkipListSet(), state);
369   }
370 
371   /*
372    * @param set
373    * @param state Accumulates deletes and candidates.
374    */
375   private void getRowKeyAtOrBefore(final NavigableSet<Cell> set,
376       final GetClosestRowBeforeTracker state) {
377     if (set.isEmpty()) {
378       return;
379     }
380     if (!walkForwardInSingleRow(set, state.getTargetKey(), state)) {
381       // Found nothing in row.  Try backing up.
382       getRowKeyBefore(set, state);
383     }
384   }
385 
386   /*
387    * Walk forward in a row from <code>firstOnRow</code>.  Presumption is that
388    * we have been passed the first possible key on a row.  As we walk forward
389    * we accumulate deletes until we hit a candidate on the row at which point
390    * we return.
391    * @param set
392    * @param firstOnRow First possible key on this row.
393    * @param state
394    * @return True if we found a candidate walking this row.
395    */
396   private boolean walkForwardInSingleRow(final SortedSet<Cell> set,
397       final Cell firstOnRow, final GetClosestRowBeforeTracker state) {
398     boolean foundCandidate = false;
399     SortedSet<Cell> tail = set.tailSet(firstOnRow);
400     if (tail.isEmpty()) return foundCandidate;
401     for (Iterator<Cell> i = tail.iterator(); i.hasNext();) {
402       Cell kv = i.next();
403       // Did we go beyond the target row? If so break.
404       if (state.isTooFar(kv, firstOnRow)) break;
405       if (state.isExpired(kv)) {
406         i.remove();
407         continue;
408       }
409       // If we added something, this row is a contender. break.
410       if (state.handle(kv)) {
411         foundCandidate = true;
412         break;
413       }
414     }
415     return foundCandidate;
416   }
417 
418   /*
419    * Walk backwards through the passed set a row at a time until we run out of
420    * set or until we get a candidate.
421    * @param set
422    * @param state
423    */
424   private void getRowKeyBefore(NavigableSet<Cell> set,
425       final GetClosestRowBeforeTracker state) {
426     Cell firstOnRow = state.getTargetKey();
427     for (Member p = memberOfPreviousRow(set, state, firstOnRow);
428         p != null; p = memberOfPreviousRow(p.set, state, firstOnRow)) {
429       // Make sure we don't fall out of our table.
430       if (!state.isTargetTable(p.cell)) break;
431       // Stop looking if we've exited the better candidate range.
432       if (!state.isBetterCandidate(p.cell)) break;
433       // Make into firstOnRow
434       firstOnRow = new KeyValue(p.cell.getRowArray(), p.cell.getRowOffset(), p.cell.getRowLength(),
435           HConstants.LATEST_TIMESTAMP);
436       // If we find something, break;
437       if (walkForwardInSingleRow(p.set, firstOnRow, state)) break;
438     }
439   }
440 
441   /**
442    * Only used by tests. TODO: Remove
443    *
444    * Given the specs of a column, update it, first by inserting a new record,
445    * then removing the old one.  Since there is only 1 KeyValue involved, the memstoreTS
446    * will be set to 0, thus ensuring that they instantly appear to anyone. The underlying
447    * store will ensure that the insert/delete each are atomic. A scanner/reader will either
448    * get the new value, or the old value and all readers will eventually only see the new
449    * value after the old was removed.
450    *
451    * @param row
452    * @param family
453    * @param qualifier
454    * @param newValue
455    * @param now
456    * @return  Timestamp
457    */
458   @Override
459   public long updateColumnValue(byte[] row,
460                                 byte[] family,
461                                 byte[] qualifier,
462                                 long newValue,
463                                 long now) {
464     Cell firstCell = KeyValueUtil.createFirstOnRow(row, family, qualifier);
465     // Is there a Cell in 'snapshot' with the same TS? If so, upgrade the timestamp a bit.
466     SortedSet<Cell> snSs = snapshotSection.getCellSkipListSet().tailSet(firstCell);
467     if (!snSs.isEmpty()) {
468       Cell snc = snSs.first();
469       // is there a matching Cell in the snapshot?
470       if (CellUtil.matchingRow(snc, firstCell) && CellUtil.matchingQualifier(snc, firstCell)) {
471         if (snc.getTimestamp() == now) {
472           // poop,
473           now += 1;
474         }
475       }
476     }
477 
478     // logic here: the new ts MUST be at least 'now'. But it could be larger if necessary.
479     // But the timestamp should also be max(now, mostRecentTsInMemstore)
480 
481     // so we cant add the new Cell w/o knowing what's there already, but we also
482     // want to take this chance to delete some cells. So two loops (sad)
483 
484     SortedSet<Cell> ss = activeSection.getCellSkipListSet().tailSet(firstCell);
485     for (Cell cell : ss) {
486       // if this isnt the row we are interested in, then bail:
487       if (!CellUtil.matchingColumn(cell, family, qualifier)
488           || !CellUtil.matchingRow(cell, firstCell)) {
489         break; // rows dont match, bail.
490       }
491 
492       // if the qualifier matches and it's a put, just RM it out of the cellSet.
493       if (cell.getTypeByte() == KeyValue.Type.Put.getCode() &&
494           cell.getTimestamp() > now && CellUtil.matchingQualifier(firstCell, cell)) {
495         now = cell.getTimestamp();
496       }
497     }
498 
499     // create or update (upsert) a new Cell with
500     // 'now' and a 0 memstoreTS == immediately visible
501     List<Cell> cells = new ArrayList<Cell>(1);
502     cells.add(new KeyValue(row, family, qualifier, now, Bytes.toBytes(newValue)));
503     return upsert(cells, 1L, null);
504   }
505 
506   /**
507    * Update or insert the specified KeyValues.
508    * <p>
509    * For each KeyValue, insert into MemStore.  This will atomically upsert the
510    * value for that row/family/qualifier.  If a KeyValue did already exist,
511    * it will then be removed.
512    * <p>
513    * Currently the memstoreTS is kept at 0 so as each insert happens, it will
514    * be immediately visible.  May want to change this so it is atomic across
515    * all KeyValues.
516    * <p>
517    * This is called under row lock, so Get operations will still see updates
518    * atomically.  Scans will only see each KeyValue update as atomic.
519    *
520    * @param cells
521    * @param readpoint readpoint below which we can safely remove duplicate KVs
522    * @param removedCells collect the removed cells. It can be null.
523    * @return change in memstore size
524    */
525   @Override
526   public long upsert(Iterable<Cell> cells, long readpoint, List<Cell> removedCells) {
527     long size = 0;
528     for (Cell cell : cells) {
529       size += upsert(cell, readpoint, removedCells);
530     }
531     return size;
532   }
533 
534   /**
535    * Inserts the specified KeyValue into MemStore and deletes any existing
536    * versions of the same row/family/qualifier as the specified KeyValue.
537    * <p>
538    * First, the specified KeyValue is inserted into the Memstore.
539    * <p>
540    * If there are any existing KeyValues in this MemStore with the same row,
541    * family, and qualifier, they are removed.
542    * <p>
543    * Callers must hold the read lock.
544    *
545    * @param cell
546    * @return change in size of MemStore
547    * @param removedCells collect the removed cells
548    */
549   private long upsert(Cell cell, long readpoint, List<Cell> removedCells) {
550     // Add the Cell to the MemStore
551     // Use the internalAdd method here since we (a) already have a lock
552     // and (b) cannot safely use the MSLAB here without potentially
553     // hitting OOME - see TestMemStore.testUpsertMSLAB for a
554     // test that triggers the pathological case if we don't avoid MSLAB
555     // here.
556     long addedSize = internalAdd(cell, false);
557 
558     // Get the Cells for the row/family/qualifier regardless of timestamp.
559     // For this case we want to clean up any other puts
560     Cell firstCell = KeyValueUtil.createFirstOnRow(
561         cell.getRowArray(), cell.getRowOffset(), cell.getRowLength(),
562         cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength(),
563         cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
564     SortedSet<Cell> ss = activeSection.getCellSkipListSet().tailSet(firstCell);
565     Iterator<Cell> it = ss.iterator();
566     // versions visible to oldest scanner
567     int versionsVisible = 0;
568     while ( it.hasNext() ) {
569       Cell cur = it.next();
570 
571       if (cell == cur) {
572         // ignore the one just put in
573         continue;
574       }
575       // check that this is the row and column we are interested in, otherwise bail
576       if (CellUtil.matchingRow(cell, cur) && CellUtil.matchingQualifier(cell, cur)) {
577         // only remove Puts that concurrent scanners cannot possibly see
578         if (cur.getTypeByte() == KeyValue.Type.Put.getCode() &&
579             cur.getSequenceId() <= readpoint) {
580           if (versionsVisible >= 1) {
581             // if we get here we have seen at least one version visible to the oldest scanner,
582             // which means we can prove that no scanner will see this version
583 
584             // false means there was a change, so give us the size.
585             long delta = heapSizeChange(cur, true);
586             addedSize -= delta;
587             activeSection.getHeapSize().addAndGet(-delta);
588             activeSection.getCellsCount().decrementAndGet();
589             if (removedCells != null) {
590               removedCells.add(cur);
591             }
592             it.remove();
593             setOldestEditTimeToNow();
594           } else {
595             versionsVisible++;
596           }
597         }
598       } else {
599         // past the row or column, done
600         break;
601       }
602     }
603     return addedSize;
604   }
605 
606   /*
607    * Immutable data structure to hold member found in set and the set it was
608    * found in. Include set because it is carrying context.
609    */
610   private static class Member {
611     final Cell cell;
612     final NavigableSet<Cell> set;
613     Member(final NavigableSet<Cell> s, final Cell kv) {
614       this.cell = kv;
615       this.set = s;
616     }
617   }
618 
619   /*
620    * @param set Set to walk back in.  Pass a first in row or we'll return
621    * same row (loop).
622    * @param state Utility and context.
623    * @param firstOnRow First item on the row after the one we want to find a
624    * member in.
625    * @return Null or member of row previous to <code>firstOnRow</code>
626    */
627   private Member memberOfPreviousRow(NavigableSet<Cell> set,
628       final GetClosestRowBeforeTracker state, final Cell firstOnRow) {
629     NavigableSet<Cell> head = set.headSet(firstOnRow, false);
630     if (head.isEmpty()) return null;
631     for (Iterator<Cell> i = head.descendingIterator(); i.hasNext();) {
632       Cell found = i.next();
633       if (state.isExpired(found)) {
634         i.remove();
635         continue;
636       }
637       return new Member(head, found);
638     }
639     return null;
640   }
641 
642   /**
643    * @return scanner on memstore and snapshot in this order.
644    */
645   @Override
646   public List<KeyValueScanner> getScanners(long readPt) {
647     MemStoreScanner scanner =
648       new MemStoreScanner(activeSection, snapshotSection, readPt, comparator);
649     scanner.seek(CellUtil.createCell(HConstants.EMPTY_START_ROW));
650     if (scanner.peek() == null) {
651       scanner.close();
652       return null;
653     }
654     return Collections.<KeyValueScanner> singletonList(scanner);
655   }
656 
657   /**
658    * Check if this memstore may contain the required keys
659    * @param scan scan
660    * @param store holds reference to cf
661    * @param oldestUnexpiredTS
662    * @return False if the key definitely does not exist in this Memstore
663    */
664   public boolean shouldSeek(Scan scan, Store store, long oldestUnexpiredTS) {
665     return shouldSeek(activeSection.getTimeRangeTracker(),
666         snapshotSection.getTimeRangeTracker(), scan, store, oldestUnexpiredTS);
667   }
668 
669   /**
670    * Check if this memstore may contain the required keys
671    * @param activeTimeRangeTracker the tracker of active data
672    * @param snapshotTimeRangeTracker the tracker of snapshot data
673    * @param scan scan
674    * @param store holds reference to cf
675    * @param oldestUnexpiredTS
676    * @return False if the key definitely does not exist in this Memstore
677    */
678   private static boolean shouldSeek(TimeRangeTracker activeTimeRangeTracker,
679       TimeRangeTracker snapshotTimeRangeTracker, Scan scan, Store store, long oldestUnexpiredTS) {
680     byte[] cf = store.getFamily().getName();
681     TimeRange timeRange = scan.getColumnFamilyTimeRange().get(cf);
682     if (timeRange == null) {
683       timeRange = scan.getTimeRange();
684     }
685     return (activeTimeRangeTracker.includesTimeRange(timeRange) ||
686       snapshotTimeRangeTracker.includesTimeRange(timeRange)) &&
687       (Math.max(activeTimeRangeTracker.getMax(), snapshotTimeRangeTracker.getMax()) >= oldestUnexpiredTS);
688   }
689 
690   /*
691    * MemStoreScanner implements the KeyValueScanner.
692    * It lets the caller scan the contents of a memstore -- both current
693    * map and snapshot.
694    * This behaves as if it were a real scanner but does not maintain position.
695    */
696   protected static class MemStoreScanner extends NonLazyKeyValueScanner {
697     // Next row information for either cellSet or snapshot
698     private Cell cellSetNextRow = null;
699     private Cell snapshotNextRow = null;
700 
701     // last iterated Cells for cellSet and snapshot (to restore iterator state after reseek)
702     private Cell cellSetItRow = null;
703     private Cell snapshotItRow = null;
704     
705     // iterator based scanning.
706     private Iterator<Cell> cellSetIt;
707     private Iterator<Cell> snapshotIt;
708 
709     // The cellSet and snapshot at the time of creating this scanner
710     private final Section activeAtCreation;
711     private final Section snapshotAtCreation;
712 
713     // the pre-calculated Cell to be returned by peek() or next()
714     private Cell theNext;
715 
716     // A flag represents whether could stop skipping Cells for MVCC
717     // if have encountered the next row. Only used for reversed scan
718     private boolean stopSkippingCellsIfNextRow = false;
719     // Stop skipping KeyValues for MVCC if finish this row. Only used for reversed scan
720     private Cell stopSkippingKVsRow;
721 
722     private final long readPoint;
723     private final KeyValue.KVComparator comparator;
724     /*
725     Some notes...
726 
727      So memstorescanner is fixed at creation time. this includes pointers/iterators into
728     existing kvset/snapshot.  during a snapshot creation, the kvset is null, and the
729     snapshot is moved.  since kvset is null there is no point on reseeking on both,
730       we can save us the trouble. During the snapshot->hfile transition, the memstore
731       scanner is re-created by StoreScanner#updateReaders().  StoreScanner should
732       potentially do something smarter by adjusting the existing memstore scanner.
733 
734       But there is a greater problem here, that being once a scanner has progressed
735       during a snapshot scenario, we currently iterate past the kvset then 'finish' up.
736       if a scan lasts a little while, there is a chance for new entries in kvset to
737       become available but we will never see them.  This needs to be handled at the
738       StoreScanner level with coordination with MemStoreScanner.
739 
740       Currently, this problem is only partly managed: during the small amount of time
741       when the StoreScanner has not yet created a new MemStoreScanner, we will miss
742       the adds to kvset in the MemStoreScanner.
743     */
744 
745     MemStoreScanner(Section activeSection, Section snapshotSection, long readPoint, final KeyValue.KVComparator c) {
746       this.readPoint = readPoint;
747       this.comparator = c;
748       activeAtCreation = activeSection;
749       snapshotAtCreation = snapshotSection;
750       if (activeAtCreation.getMemStoreLAB() != null) {
751         activeAtCreation.getMemStoreLAB().incScannerCount();
752       }
753       if (snapshotAtCreation.getMemStoreLAB() != null) {
754         snapshotAtCreation.getMemStoreLAB().incScannerCount();
755       }
756       if (Trace.isTracing() && Trace.currentSpan() != null) {
757         Trace.currentSpan().addTimelineAnnotation("Creating MemStoreScanner");
758       }
759     }
760 
761     /**
762      * Lock on 'this' must be held by caller.
763      * @param it
764      * @return Next Cell
765      */
766     private Cell getNext(Iterator<Cell> it) {
767       Cell v = null;
768       try {
769         while (it.hasNext()) {
770           v = it.next();
771           if (v.getSequenceId() <= this.readPoint) {
772             return v;
773           }
774           if (stopSkippingCellsIfNextRow && stopSkippingKVsRow != null
775               && comparator.compareRows(v, stopSkippingKVsRow) > 0) {
776             return null;
777           }
778         }
779 
780         return null;
781       } finally {
782         if (v != null) {
783           // in all cases, remember the last Cell iterated to
784           if (it == snapshotIt) {
785             snapshotItRow = v;
786           } else {
787             cellSetItRow = v;
788           }
789         }
790       }
791     }
792 
793     /**
794      *  Set the scanner at the seek key.
795      *  Must be called only once: there is no thread safety between the scanner
796      *   and the memStore.
797      * @param key seek value
798      * @return false if the key is null or if there is no data
799      */
800     @Override
801     public synchronized boolean seek(Cell key) {
802       if (key == null) {
803         close();
804         return false;
805       }
806       // kvset and snapshot will never be null.
807       // if tailSet can't find anything, SortedSet is empty (not null).
808       cellSetIt = activeAtCreation.getCellSkipListSet().tailSet(key).iterator();
809       snapshotIt = snapshotAtCreation.getCellSkipListSet().tailSet(key).iterator();
810       cellSetItRow = null;
811       snapshotItRow = null;
812 
813       return seekInSubLists(key);
814     }
815 
816 
817     /**
818      * (Re)initialize the iterators after a seek or a reseek.
819      */
820     private synchronized boolean seekInSubLists(Cell key){
821       cellSetNextRow = getNext(cellSetIt);
822       snapshotNextRow = getNext(snapshotIt);
823 
824       // Calculate the next value
825       theNext = getLowest(cellSetNextRow, snapshotNextRow);
826 
827       // has data
828       return (theNext != null);
829     }
830 
831 
832     /**
833      * Move forward on the sub-lists set previously by seek.
834      * @param key seek value (should be non-null)
835      * @return true if there is at least one KV to read, false otherwise
836      */
837     @Override
838     public synchronized boolean reseek(Cell key) {
839       /*
840       See HBASE-4195 & HBASE-3855 & HBASE-6591 for the background on this implementation.
841       This code is executed concurrently with flush and puts, without locks.
842       Two points must be known when working on this code:
843       1) It's not possible to use the 'kvTail' and 'snapshot'
844        variables, as they are modified during a flush.
845       2) The ideal implementation for performance would use the sub skip list
846        implicitly pointed by the iterators 'kvsetIt' and
847        'snapshotIt'. Unfortunately the Java API does not offer a method to
848        get it. So we remember the last keys we iterated to and restore
849        the reseeked set to at least that point.
850        */
851       cellSetIt = activeAtCreation.getCellSkipListSet().tailSet(getHighest(key, cellSetItRow)).iterator();
852       snapshotIt = snapshotAtCreation.getCellSkipListSet().tailSet(getHighest(key, snapshotItRow)).iterator();
853 
854       return seekInSubLists(key);
855     }
856 
857 
858     @Override
859     public synchronized Cell peek() {
860       //DebugPrint.println(" MS@" + hashCode() + " peek = " + getLowest());
861       return theNext;
862     }
863 
864     @Override
865     public synchronized Cell next() {
866       if (theNext == null) {
867           return null;
868       }
869 
870       final Cell ret = theNext;
871 
872       // Advance one of the iterators
873       if (theNext == cellSetNextRow) {
874         cellSetNextRow = getNext(cellSetIt);
875       } else {
876         snapshotNextRow = getNext(snapshotIt);
877       }
878 
879       // Calculate the next value
880       theNext = getLowest(cellSetNextRow, snapshotNextRow);
881 
882       //long readpoint = ReadWriteConsistencyControl.getThreadReadPoint();
883       //DebugPrint.println(" MS@" + hashCode() + " next: " + theNext + " next_next: " +
884       //    getLowest() + " threadpoint=" + readpoint);
885       return ret;
886     }
887 
888     /*
889      * Returns the lower of the two key values, or null if they are both null.
890      * This uses comparator.compare() to compare the KeyValue using the memstore
891      * comparator.
892      */
893     private Cell getLowest(Cell first, Cell second) {
894       if (first == null && second == null) {
895         return null;
896       }
897       if (first != null && second != null) {
898         int compare = comparator.compare(first, second);
899         return (compare <= 0 ? first : second);
900       }
901       return (first != null ? first : second);
902     }
903 
904     /*
905      * Returns the higher of the two cells, or null if they are both null.
906      * This uses comparator.compare() to compare the Cell using the memstore
907      * comparator.
908      */
909     private Cell getHighest(Cell first, Cell second) {
910       if (first == null && second == null) {
911         return null;
912       }
913       if (first != null && second != null) {
914         int compare = comparator.compare(first, second);
915         return (compare > 0 ? first : second);
916       }
917       return (first != null ? first : second);
918     }
919 
920     @Override
921     public synchronized void close() {
922       this.cellSetNextRow = null;
923       this.snapshotNextRow = null;
924 
925       this.cellSetIt = null;
926       this.snapshotIt = null;
927 
928       if (activeAtCreation != null && activeAtCreation.getMemStoreLAB() != null) {
929         activeAtCreation.getMemStoreLAB().decScannerCount();
930       }
931       if (snapshotAtCreation != null && snapshotAtCreation.getMemStoreLAB() != null) {
932         snapshotAtCreation.getMemStoreLAB().decScannerCount();
933       }
934 
935       this.cellSetItRow = null;
936       this.snapshotItRow = null;
937     }
938 
939     /**
940      * MemStoreScanner returns max value as sequence id because it will
941      * always have the latest data among all files.
942      */
943     @Override
944     public long getSequenceID() {
945       return Long.MAX_VALUE;
946     }
947 
948     @Override
949     public boolean shouldUseScanner(Scan scan, Store store, long oldestUnexpiredTS) {
950       return shouldSeek(activeAtCreation.getTimeRangeTracker(),
951         snapshotAtCreation.getTimeRangeTracker(), scan, store, oldestUnexpiredTS);
952     }
953 
954     /**
955      * Seek scanner to the given key first. If it returns false(means
956      * peek()==null) or scanner's peek row is bigger than row of given key, seek
957      * the scanner to the previous row of given key
958      */
959     @Override
960     public synchronized boolean backwardSeek(Cell key) {
961       seek(key);
962       if (peek() == null || comparator.compareRows(peek(), key) > 0) {
963         return seekToPreviousRow(key);
964       }
965       return true;
966     }
967 
968     /**
969      * Separately get the KeyValue before the specified key from kvset and
970      * snapshotset, and use the row of higher one as the previous row of
971      * specified key, then seek to the first KeyValue of previous row
972      */
973     @Override
974     public synchronized boolean seekToPreviousRow(Cell originalKey) {
975       boolean keepSeeking = false;
976       Cell key = originalKey;
977       do {
978         Cell firstKeyOnRow = KeyValueUtil.createFirstOnRow(key.getRowArray(), key.getRowOffset(),
979             key.getRowLength());
980         SortedSet<Cell> cellHead = activeAtCreation.getCellSkipListSet().headSet(firstKeyOnRow);
981         Cell cellSetBeforeRow = cellHead.isEmpty() ? null : cellHead.last();
982         SortedSet<Cell> snapshotHead = snapshotAtCreation.getCellSkipListSet()
983             .headSet(firstKeyOnRow);
984         Cell snapshotBeforeRow = snapshotHead.isEmpty() ? null : snapshotHead
985             .last();
986         Cell lastCellBeforeRow = getHighest(cellSetBeforeRow, snapshotBeforeRow);
987         if (lastCellBeforeRow == null) {
988           theNext = null;
989           return false;
990         }
991         Cell firstKeyOnPreviousRow = KeyValueUtil.createFirstOnRow(lastCellBeforeRow.getRowArray(),
992             lastCellBeforeRow.getRowOffset(), lastCellBeforeRow.getRowLength());
993         this.stopSkippingCellsIfNextRow = true;
994         this.stopSkippingKVsRow = firstKeyOnPreviousRow;
995         seek(firstKeyOnPreviousRow);
996         this.stopSkippingCellsIfNextRow = false;
997         if (peek() == null
998             || comparator.compareRows(peek(), firstKeyOnPreviousRow) > 0) {
999           keepSeeking = true;
1000           key = firstKeyOnPreviousRow;
1001           continue;
1002         } else {
1003           keepSeeking = false;
1004         }
1005       } while (keepSeeking);
1006       return true;
1007     }
1008 
1009     @Override
1010     public synchronized boolean seekToLastRow() {
1011       Cell first = activeAtCreation.getCellSkipListSet().isEmpty() ? null
1012           : activeAtCreation.getCellSkipListSet().last();
1013       Cell second = snapshotAtCreation.getCellSkipListSet().isEmpty() ? null
1014           : snapshotAtCreation.getCellSkipListSet().last();
1015       Cell higherCell = getHighest(first, second);
1016       if (higherCell == null) {
1017         return false;
1018       }
1019       Cell firstCellOnLastRow = KeyValueUtil.createFirstOnRow(higherCell.getRowArray(),
1020           higherCell.getRowOffset(), higherCell.getRowLength());
1021       if (seek(firstCellOnLastRow)) {
1022         return true;
1023       } else {
1024         return seekToPreviousRow(higherCell);
1025       }
1026 
1027     }
1028   }
1029 
1030   public final static long FIXED_OVERHEAD = ClassSize.align(ClassSize.OBJECT
1031       + (4 * ClassSize.REFERENCE) + (2 * Bytes.SIZEOF_LONG) + Bytes.SIZEOF_BOOLEAN);
1032 
1033   public final static long DEEP_OVERHEAD = ClassSize.align(FIXED_OVERHEAD +
1034       (2 * ClassSize.ATOMIC_LONG) + (2 * ClassSize.TIMERANGE_TRACKER) +
1035       (2 * ClassSize.CELL_SKIPLIST_SET) + (2 * ClassSize.CONCURRENT_SKIPLISTMAP) +
1036       ClassSize.ATOMIC_INTEGER);
1037 
1038   /*
1039    * Calculate how the MemStore size has changed.  Includes overhead of the
1040    * backing Map.
1041    * @param cell
1042    * @param notpresent True if the cell was NOT present in the set.
1043    * @return Size
1044    */
1045   static long heapSizeChange(final Cell cell, final boolean notpresent) {
1046     return notpresent ? ClassSize.align(ClassSize.CONCURRENT_SKIPLISTMAP_ENTRY
1047         + CellUtil.estimatedHeapSizeOf(cell)) : 0;
1048   }
1049 
1050   private long keySize() {
1051     return heapSize() - DEEP_OVERHEAD;
1052   }
1053 
1054   /**
1055    * Get the entire heap usage for this MemStore not including keys in the
1056    * snapshot.
1057    */
1058   @Override
1059   public long heapSize() {
1060     return activeSection.getHeapSize().get();
1061   }
1062 
1063   @Override
1064   public long size() {
1065     return heapSize();
1066   }
1067 
1068   /**
1069    * Code to help figure if our approximation of object heap sizes is close
1070    * enough.  See hbase-900.  Fills memstores then waits so user can heap
1071    * dump and bring up resultant hprof in something like jprofiler which
1072    * allows you get 'deep size' on objects.
1073    * @param args main args
1074    */
1075   public static void main(String [] args) {
1076     RuntimeMXBean runtime = ManagementFactory.getRuntimeMXBean();
1077     LOG.info("vmName=" + runtime.getVmName() + ", vmVendor=" +
1078       runtime.getVmVendor() + ", vmVersion=" + runtime.getVmVersion());
1079     LOG.info("vmInputArguments=" + runtime.getInputArguments());
1080     DefaultMemStore memstore1 = new DefaultMemStore();
1081     // TODO: x32 vs x64
1082     long size = 0;
1083     final int count = 10000;
1084     byte [] fam = Bytes.toBytes("col");
1085     byte [] qf = Bytes.toBytes("umn");
1086     byte [] empty = new byte[0];
1087     for (int i = 0; i < count; i++) {
1088       // Give each its own ts
1089       size += memstore1.add(new KeyValue(Bytes.toBytes(i), fam, qf, i, empty));
1090     }
1091     LOG.info("memstore1 estimated size=" + size);
1092     for (int i = 0; i < count; i++) {
1093       size += memstore1.add(new KeyValue(Bytes.toBytes(i), fam, qf, i, empty));
1094     }
1095     LOG.info("memstore1 estimated size (2nd loading of same data)=" + size);
1096     // Make a variably sized memstore.
1097     DefaultMemStore memstore2 = new DefaultMemStore();
1098     for (int i = 0; i < count; i++) {
1099       size += memstore2.add(new KeyValue(Bytes.toBytes(i), fam, qf, i, new byte[i]));
1100     }
1101     LOG.info("memstore2 estimated size=" + size);
1102     final int seconds = 30;
1103     LOG.info("Waiting " + seconds + " seconds while heap dump is taken");
1104     for (int i = 0; i < seconds; i++) {
1105       // Thread.sleep(1000);
1106     }
1107     LOG.info("Exiting.");
1108   }
1109 
1110   /**
1111    * Contains the fields which are useful to MemStoreScanner.
1112    */
1113   @InterfaceAudience.Private
1114   @VisibleForTesting
1115   static class Section {
1116     /**
1117      * MemStore.  Use a CellSkipListSet rather than SkipListSet because of the
1118      * better semantics.  The Map will overwrite if passed a key it already had
1119      * whereas the Set will not add new Cell if key is same though value might be
1120      * different.  Value is not important -- just make sure always same reference passed.
1121      */
1122     private final CellSkipListSet cellSet;
1123     private final TimeRangeTracker tracker = new TimeRangeTracker();
1124     /**
1125      * Used to track own heapSize.
1126      */
1127     private final AtomicLong heapSize;
1128     private final AtomicInteger cellCount;
1129     private final MemStoreLAB allocator;
1130 
1131     static Section newSnapshotSection(final KeyValue.KVComparator c) {
1132       return new Section(c, null, 0);
1133     }
1134 
1135     static Section newActiveSection(final KeyValue.KVComparator c,
1136             final Configuration conf) {
1137       return new Section(c, conf, DEEP_OVERHEAD);
1138     }
1139 
1140     private Section(final KeyValue.KVComparator c,
1141             final Configuration conf, long initHeapSize) {
1142       this.cellSet = new CellSkipListSet(c);
1143       this.heapSize = new AtomicLong(initHeapSize);
1144       this.cellCount = new AtomicInteger(0);
1145       if (conf != null && conf.getBoolean(USEMSLAB_KEY, USEMSLAB_DEFAULT)) {
1146         String className = conf.get(MSLAB_CLASS_NAME, HeapMemStoreLAB.class.getName());
1147         this.allocator = ReflectionUtils.instantiateWithCustomCtor(className,
1148                 new Class[]{Configuration.class}, new Object[]{conf});
1149       } else {
1150         this.allocator = null;
1151       }
1152     }
1153 
1154     CellSkipListSet getCellSkipListSet() {
1155       return cellSet;
1156     }
1157 
1158     TimeRangeTracker getTimeRangeTracker() {
1159       return tracker;
1160     }
1161 
1162     AtomicLong getHeapSize() {
1163       return heapSize;
1164     }
1165 
1166     AtomicInteger getCellsCount() {
1167       return cellCount;
1168     }
1169 
1170     MemStoreLAB getMemStoreLAB() {
1171       return allocator;
1172     }
1173   }
1174 }