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