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 
723     private final long readPoint;
724     private final KeyValue.KVComparator comparator;
725     /*
726     Some notes...
727 
728      So memstorescanner is fixed at creation time. this includes pointers/iterators into
729     existing kvset/snapshot.  during a snapshot creation, the kvset is null, and the
730     snapshot is moved.  since kvset is null there is no point on reseeking on both,
731       we can save us the trouble. During the snapshot->hfile transition, the memstore
732       scanner is re-created by StoreScanner#updateReaders().  StoreScanner should
733       potentially do something smarter by adjusting the existing memstore scanner.
734 
735       But there is a greater problem here, that being once a scanner has progressed
736       during a snapshot scenario, we currently iterate past the kvset then 'finish' up.
737       if a scan lasts a little while, there is a chance for new entries in kvset to
738       become available but we will never see them.  This needs to be handled at the
739       StoreScanner level with coordination with MemStoreScanner.
740 
741       Currently, this problem is only partly managed: during the small amount of time
742       when the StoreScanner has not yet created a new MemStoreScanner, we will miss
743       the adds to kvset in the MemStoreScanner.
744     */
745 
746     MemStoreScanner(Section activeSection, Section snapshotSection, long readPoint, final KeyValue.KVComparator c) {
747       this.readPoint = readPoint;
748       this.comparator = c;
749       activeAtCreation = activeSection;
750       snapshotAtCreation = snapshotSection;
751       if (activeAtCreation.getMemStoreLAB() != null) {
752         activeAtCreation.getMemStoreLAB().incScannerCount();
753       }
754       if (snapshotAtCreation.getMemStoreLAB() != null) {
755         snapshotAtCreation.getMemStoreLAB().incScannerCount();
756       }
757       if (Trace.isTracing() && Trace.currentSpan() != null) {
758         Trace.currentSpan().addTimelineAnnotation("Creating MemStoreScanner");
759       }
760     }
761 
762     /**
763      * Lock on 'this' must be held by caller.
764      * @param it
765      * @return Next Cell
766      */
767     private Cell getNext(Iterator<Cell> it) {
768       Cell startCell = theNext;
769       Cell v = null;
770       try {
771         while (it.hasNext()) {
772           v = it.next();
773           if (v.getSequenceId() <= this.readPoint) {
774             return v;
775           }
776           if (stopSkippingCellsIfNextRow && startCell != null
777               && comparator.compareRows(v, startCell) > 0) {
778             return null;
779           }
780         }
781 
782         return null;
783       } finally {
784         if (v != null) {
785           // in all cases, remember the last Cell iterated to
786           if (it == snapshotIt) {
787             snapshotItRow = v;
788           } else {
789             cellSetItRow = v;
790           }
791         }
792       }
793     }
794 
795     /**
796      *  Set the scanner at the seek key.
797      *  Must be called only once: there is no thread safety between the scanner
798      *   and the memStore.
799      * @param key seek value
800      * @return false if the key is null or if there is no data
801      */
802     @Override
803     public synchronized boolean seek(Cell key) {
804       if (key == null) {
805         close();
806         return false;
807       }
808       // kvset and snapshot will never be null.
809       // if tailSet can't find anything, SortedSet is empty (not null).
810       cellSetIt = activeAtCreation.getCellSkipListSet().tailSet(key).iterator();
811       snapshotIt = snapshotAtCreation.getCellSkipListSet().tailSet(key).iterator();
812       cellSetItRow = null;
813       snapshotItRow = null;
814 
815       return seekInSubLists(key);
816     }
817 
818 
819     /**
820      * (Re)initialize the iterators after a seek or a reseek.
821      */
822     private synchronized boolean seekInSubLists(Cell key){
823       cellSetNextRow = getNext(cellSetIt);
824       snapshotNextRow = getNext(snapshotIt);
825 
826       // Calculate the next value
827       theNext = getLowest(cellSetNextRow, snapshotNextRow);
828 
829       // has data
830       return (theNext != null);
831     }
832 
833 
834     /**
835      * Move forward on the sub-lists set previously by seek.
836      * @param key seek value (should be non-null)
837      * @return true if there is at least one KV to read, false otherwise
838      */
839     @Override
840     public synchronized boolean reseek(Cell key) {
841       /*
842       See HBASE-4195 & HBASE-3855 & HBASE-6591 for the background on this implementation.
843       This code is executed concurrently with flush and puts, without locks.
844       Two points must be known when working on this code:
845       1) It's not possible to use the 'kvTail' and 'snapshot'
846        variables, as they are modified during a flush.
847       2) The ideal implementation for performance would use the sub skip list
848        implicitly pointed by the iterators 'kvsetIt' and
849        'snapshotIt'. Unfortunately the Java API does not offer a method to
850        get it. So we remember the last keys we iterated to and restore
851        the reseeked set to at least that point.
852        */
853       cellSetIt = activeAtCreation.getCellSkipListSet().tailSet(getHighest(key, cellSetItRow)).iterator();
854       snapshotIt = snapshotAtCreation.getCellSkipListSet().tailSet(getHighest(key, snapshotItRow)).iterator();
855 
856       return seekInSubLists(key);
857     }
858 
859 
860     @Override
861     public synchronized Cell peek() {
862       //DebugPrint.println(" MS@" + hashCode() + " peek = " + getLowest());
863       return theNext;
864     }
865 
866     @Override
867     public synchronized Cell next() {
868       if (theNext == null) {
869           return null;
870       }
871 
872       final Cell ret = theNext;
873 
874       // Advance one of the iterators
875       if (theNext == cellSetNextRow) {
876         cellSetNextRow = getNext(cellSetIt);
877       } else {
878         snapshotNextRow = getNext(snapshotIt);
879       }
880 
881       // Calculate the next value
882       theNext = getLowest(cellSetNextRow, snapshotNextRow);
883 
884       //long readpoint = ReadWriteConsistencyControl.getThreadReadPoint();
885       //DebugPrint.println(" MS@" + hashCode() + " next: " + theNext + " next_next: " +
886       //    getLowest() + " threadpoint=" + readpoint);
887       return ret;
888     }
889 
890     /*
891      * Returns the lower of the two key values, or null if they are both null.
892      * This uses comparator.compare() to compare the KeyValue using the memstore
893      * comparator.
894      */
895     private Cell getLowest(Cell first, Cell second) {
896       if (first == null && second == null) {
897         return null;
898       }
899       if (first != null && second != null) {
900         int compare = comparator.compare(first, second);
901         return (compare <= 0 ? first : second);
902       }
903       return (first != null ? first : second);
904     }
905 
906     /*
907      * Returns the higher of the two cells, or null if they are both null.
908      * This uses comparator.compare() to compare the Cell using the memstore
909      * comparator.
910      */
911     private Cell getHighest(Cell first, Cell second) {
912       if (first == null && second == null) {
913         return null;
914       }
915       if (first != null && second != null) {
916         int compare = comparator.compare(first, second);
917         return (compare > 0 ? first : second);
918       }
919       return (first != null ? first : second);
920     }
921 
922     @Override
923     public synchronized void close() {
924       this.cellSetNextRow = null;
925       this.snapshotNextRow = null;
926 
927       this.cellSetIt = null;
928       this.snapshotIt = null;
929 
930       if (activeAtCreation != null && activeAtCreation.getMemStoreLAB() != null) {
931         activeAtCreation.getMemStoreLAB().decScannerCount();
932       }
933       if (snapshotAtCreation != null && snapshotAtCreation.getMemStoreLAB() != null) {
934         snapshotAtCreation.getMemStoreLAB().decScannerCount();
935       }
936 
937       this.cellSetItRow = null;
938       this.snapshotItRow = null;
939     }
940 
941     /**
942      * MemStoreScanner returns Long.MAX_VALUE because it will always have the latest data among all
943      * scanners.
944      * @see KeyValueScanner#getScannerOrder()
945      */
946     @Override
947     public long getScannerOrder() {
948       return Long.MAX_VALUE;
949     }
950 
951     @Override
952     public boolean shouldUseScanner(Scan scan, Store store, long oldestUnexpiredTS) {
953       return shouldSeek(activeAtCreation.getTimeRangeTracker(),
954         snapshotAtCreation.getTimeRangeTracker(), scan, store, oldestUnexpiredTS);
955     }
956 
957     /**
958      * Seek scanner to the given key first. If it returns false(means
959      * peek()==null) or scanner's peek row is bigger than row of given key, seek
960      * the scanner to the previous row of given key
961      */
962     @Override
963     public synchronized boolean backwardSeek(Cell key) {
964       seek(key);
965       if (peek() == null || comparator.compareRows(peek(), key) > 0) {
966         return seekToPreviousRow(key);
967       }
968       return true;
969     }
970 
971     /**
972      * Separately get the KeyValue before the specified key from kvset and
973      * snapshotset, and use the row of higher one as the previous row of
974      * specified key, then seek to the first KeyValue of previous row
975      */
976     @Override
977     public synchronized boolean seekToPreviousRow(Cell originalKey) {
978       boolean keepSeeking = false;
979       Cell key = originalKey;
980       do {
981         Cell firstKeyOnRow = KeyValueUtil.createFirstOnRow(key.getRowArray(), key.getRowOffset(),
982             key.getRowLength());
983         SortedSet<Cell> cellHead = activeAtCreation.getCellSkipListSet().headSet(firstKeyOnRow);
984         Cell cellSetBeforeRow = cellHead.isEmpty() ? null : cellHead.last();
985         SortedSet<Cell> snapshotHead = snapshotAtCreation.getCellSkipListSet()
986             .headSet(firstKeyOnRow);
987         Cell snapshotBeforeRow = snapshotHead.isEmpty() ? null : snapshotHead
988             .last();
989         Cell lastCellBeforeRow = getHighest(cellSetBeforeRow, snapshotBeforeRow);
990         if (lastCellBeforeRow == null) {
991           theNext = null;
992           return false;
993         }
994         Cell firstKeyOnPreviousRow = KeyValueUtil.createFirstOnRow(lastCellBeforeRow.getRowArray(),
995             lastCellBeforeRow.getRowOffset(), lastCellBeforeRow.getRowLength());
996         this.stopSkippingCellsIfNextRow = true;
997         seek(firstKeyOnPreviousRow);
998         this.stopSkippingCellsIfNextRow = false;
999         if (peek() == null
1000             || comparator.compareRows(peek(), firstKeyOnPreviousRow) > 0) {
1001           keepSeeking = true;
1002           key = firstKeyOnPreviousRow;
1003           continue;
1004         } else {
1005           keepSeeking = false;
1006         }
1007       } while (keepSeeking);
1008       return true;
1009     }
1010 
1011     @Override
1012     public synchronized boolean seekToLastRow() {
1013       Cell first = activeAtCreation.getCellSkipListSet().isEmpty() ? null
1014         : activeAtCreation.getCellSkipListSet().last();
1015       Cell second = snapshotAtCreation.getCellSkipListSet().isEmpty() ? null
1016           : snapshotAtCreation.getCellSkipListSet().last();
1017       Cell higherCell = getHighest(first, second);
1018       if (higherCell == null) {
1019         return false;
1020       }
1021       Cell firstCellOnLastRow = KeyValueUtil.createFirstOnRow(higherCell.getRowArray(),
1022           higherCell.getRowOffset(), higherCell.getRowLength());
1023       if (seek(firstCellOnLastRow)) {
1024         return true;
1025       } else {
1026         return seekToPreviousRow(higherCell);
1027       }
1028 
1029     }
1030   }
1031 
1032   public final static long FIXED_OVERHEAD = ClassSize.align(ClassSize.OBJECT
1033       + (4 * ClassSize.REFERENCE) + (2 * Bytes.SIZEOF_LONG) + Bytes.SIZEOF_BOOLEAN);
1034 
1035   public final static long DEEP_OVERHEAD = ClassSize.align(FIXED_OVERHEAD +
1036       (2 * ClassSize.ATOMIC_LONG) + (2 * ClassSize.TIMERANGE_TRACKER) +
1037       (2 * ClassSize.CELL_SKIPLIST_SET) + (2 * ClassSize.CONCURRENT_SKIPLISTMAP));
1038 
1039   /*
1040    * Calculate how the MemStore size has changed.  Includes overhead of the
1041    * backing Map.
1042    * @param cell
1043    * @param notpresent True if the cell was NOT present in the set.
1044    * @return Size
1045    */
1046   static long heapSizeChange(final Cell cell, final boolean notpresent) {
1047     return notpresent ? ClassSize.align(ClassSize.CONCURRENT_SKIPLISTMAP_ENTRY
1048         + CellUtil.estimatedHeapSizeOf(cell)) : 0;
1049   }
1050 
1051   private long keySize() {
1052     return heapSize() - DEEP_OVERHEAD;
1053   }
1054 
1055   /**
1056    * Get the entire heap usage for this MemStore not including keys in the
1057    * snapshot.
1058    */
1059   @Override
1060   public long heapSize() {
1061     return activeSection.getHeapSize().get();
1062   }
1063 
1064   @Override
1065   public long size() {
1066     return heapSize();
1067   }
1068 
1069   /**
1070    * Code to help figure if our approximation of object heap sizes is close
1071    * enough.  See hbase-900.  Fills memstores then waits so user can heap
1072    * dump and bring up resultant hprof in something like jprofiler which
1073    * allows you get 'deep size' on objects.
1074    * @param args main args
1075    */
1076   public static void main(String [] args) {
1077     RuntimeMXBean runtime = ManagementFactory.getRuntimeMXBean();
1078     LOG.info("vmName=" + runtime.getVmName() + ", vmVendor=" +
1079       runtime.getVmVendor() + ", vmVersion=" + runtime.getVmVersion());
1080     LOG.info("vmInputArguments=" + runtime.getInputArguments());
1081     DefaultMemStore memstore1 = new DefaultMemStore();
1082     // TODO: x32 vs x64
1083     long size = 0;
1084     final int count = 10000;
1085     byte [] fam = Bytes.toBytes("col");
1086     byte [] qf = Bytes.toBytes("umn");
1087     byte [] empty = new byte[0];
1088     for (int i = 0; i < count; i++) {
1089       // Give each its own ts
1090       size += memstore1.add(new KeyValue(Bytes.toBytes(i), fam, qf, i, empty));
1091     }
1092     LOG.info("memstore1 estimated size=" + size);
1093     for (int i = 0; i < count; i++) {
1094       size += memstore1.add(new KeyValue(Bytes.toBytes(i), fam, qf, i, empty));
1095     }
1096     LOG.info("memstore1 estimated size (2nd loading of same data)=" + size);
1097     // Make a variably sized memstore.
1098     DefaultMemStore memstore2 = new DefaultMemStore();
1099     for (int i = 0; i < count; i++) {
1100       size += memstore2.add(new KeyValue(Bytes.toBytes(i), fam, qf, i, new byte[i]));
1101     }
1102     LOG.info("memstore2 estimated size=" + size);
1103     final int seconds = 30;
1104     LOG.info("Waiting " + seconds + " seconds while heap dump is taken");
1105     for (int i = 0; i < seconds; i++) {
1106       // Thread.sleep(1000);
1107     }
1108     LOG.info("Exiting.");
1109   }
1110 
1111   /**
1112    * Contains the fields which are useful to MemStoreScanner.
1113    */
1114   @InterfaceAudience.Private
1115   @VisibleForTesting
1116   static class Section {
1117     /**
1118      * MemStore.  Use a CellSkipListSet rather than SkipListSet because of the
1119      * better semantics.  The Map will overwrite if passed a key it already had
1120      * whereas the Set will not add new Cell if key is same though value might be
1121      * different.  Value is not important -- just make sure always same reference passed.
1122      */
1123     private final CellSkipListSet cellSet;
1124     private final TimeRangeTracker tracker = new TimeRangeTracker();
1125     /**
1126      * Used to track own heapSize.
1127      */
1128     private final AtomicLong heapSize;
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       if (conf != null && conf.getBoolean(USEMSLAB_KEY, USEMSLAB_DEFAULT)) {
1145         String className = conf.get(MSLAB_CLASS_NAME, HeapMemStoreLAB.class.getName());
1146         this.allocator = ReflectionUtils.instantiateWithCustomCtor(className,
1147                 new Class[]{Configuration.class}, new Object[]{conf});
1148       } else {
1149         this.allocator = null;
1150       }
1151     }
1152 
1153     CellSkipListSet getCellSkipListSet() {
1154       return cellSet;
1155     }
1156 
1157     TimeRangeTracker getTimeRangeTracker() {
1158       return tracker;
1159     }
1160 
1161     AtomicLong getHeapSize() {
1162       return heapSize;
1163     }
1164 
1165     MemStoreLAB getMemStoreLAB() {
1166       return allocator;
1167     }
1168 
1169   }
1170 }