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