Clean up SnapshotImpl.

* Omit SnapshotImpl::list_ when assert() isn't on
* Make SnapshotImpl::number_ const and set it in the constructor
* Make SnapshotImpl::number_ private and access it via a getter
* Rename SnapshotImpl::number_ to SnapshotImpl::sequence_number_
* Rename SnapshotList::list_ to SnapshotList::head_
* Wrap casting from Snapshot* to SnapshotImpl* in ToSnapshotImpl()

-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=194852828

db/db_impl.cc

@@ -906,7 +906,7 @@ Status DBImpl::DoCompactionWork(CompactionState* compact) {
   if (snapshots_.empty()) {
     compact->smallest_snapshot = versions_->LastSequence();
   } else {
-    compact->smallest_snapshot = snapshots_.oldest()->number_;
+    compact->smallest_snapshot = snapshots_.oldest()->sequence_number();
   }
 
   // Release mutex while we're actually doing the compaction work
@@ -1121,7 +1121,8 @@ Status DBImpl::Get(const ReadOptions& options,
   MutexLock l(&mutex_);
   SequenceNumber snapshot;
   if (options.snapshot != nullptr) {
-    snapshot = reinterpret_cast<const SnapshotImpl*>(options.snapshot)->number_;
+    snapshot =
+        static_cast<const SnapshotImpl*>(options.snapshot)->sequence_number();
   } else {
     snapshot = versions_->LastSequence();
   }
@@ -1168,7 +1169,7 @@ Iterator* DBImpl::NewIterator(const ReadOptions& options) {
   return NewDBIterator(
       this, user_comparator(), iter,
       (options.snapshot != nullptr
-       ? reinterpret_cast<const SnapshotImpl*>(options.snapshot)->number_
+       ? static_cast<const SnapshotImpl*>(options.snapshot)->sequence_number()
        : latest_snapshot),
       seed);
 }
@@ -1185,9 +1186,9 @@ const Snapshot* DBImpl::GetSnapshot() {
   return snapshots_.New(versions_->LastSequence());
 }
 
-void DBImpl::ReleaseSnapshot(const Snapshot* s) {
+void DBImpl::ReleaseSnapshot(const Snapshot* snapshot) {
   MutexLock l(&mutex_);
-  snapshots_.Delete(reinterpret_cast<const SnapshotImpl*>(s));
+  snapshots_.Delete(static_cast<const SnapshotImpl*>(snapshot));
 }
 
 // Convenience methods

db/db_test.cc

@@ -631,6 +631,55 @@ TEST(DBTest, GetSnapshot) {
   } while (ChangeOptions());
 }
 
+TEST(DBTest, GetIdenticalSnapshots) {
+  do {
+    // Try with both a short key and a long key
+    for (int i = 0; i < 2; i++) {
+      std::string key = (i == 0) ? std::string("foo") : std::string(200, 'x');
+      ASSERT_OK(Put(key, "v1"));
+      const Snapshot* s1 = db_->GetSnapshot();
+      const Snapshot* s2 = db_->GetSnapshot();
+      const Snapshot* s3 = db_->GetSnapshot();
+      ASSERT_OK(Put(key, "v2"));
+      ASSERT_EQ("v2", Get(key));
+      ASSERT_EQ("v1", Get(key, s1));
+      ASSERT_EQ("v1", Get(key, s2));
+      ASSERT_EQ("v1", Get(key, s3));
+      db_->ReleaseSnapshot(s1);
+      dbfull()->TEST_CompactMemTable();
+      ASSERT_EQ("v2", Get(key));
+      ASSERT_EQ("v1", Get(key, s2));
+      db_->ReleaseSnapshot(s2);
+      ASSERT_EQ("v1", Get(key, s3));
+      db_->ReleaseSnapshot(s3);
+    }
+  } while (ChangeOptions());
+}
+
+TEST(DBTest, IterateOverEmptySnapshot) {
+  do {
+    const Snapshot* snapshot = db_->GetSnapshot();
+    ReadOptions read_options;
+    read_options.snapshot = snapshot;
+    ASSERT_OK(Put("foo", "v1"));
+    ASSERT_OK(Put("foo", "v2"));
+
+    Iterator* iterator1 = db_->NewIterator(read_options);
+    iterator1->SeekToFirst();
+    ASSERT_TRUE(!iterator1->Valid());
+    delete iterator1;
+
+    dbfull()->TEST_CompactMemTable();
+
+    Iterator* iterator2 = db_->NewIterator(read_options);
+    iterator2->SeekToFirst();
+    ASSERT_TRUE(!iterator2->Valid());
+    delete iterator2;
+
+    db_->ReleaseSnapshot(snapshot);
+  } while (ChangeOptions());
+}
+
 TEST(DBTest, GetLevel0Ordering) {
   do {
     // Check that we process level-0 files in correct order.  The code

db/snapshot.h

@@ -16,50 +16,72 @@ class SnapshotList;
 // Each SnapshotImpl corresponds to a particular sequence number.
 class SnapshotImpl : public Snapshot {
  public:
-  SequenceNumber number_;  // const after creation
+  SnapshotImpl(SequenceNumber sequence_number)
+      : sequence_number_(sequence_number) {}
+
+  SequenceNumber sequence_number() const { return sequence_number_; }
 
  private:
   friend class SnapshotList;
 
-  // SnapshotImpl is kept in a doubly-linked circular list
+  // SnapshotImpl is kept in a doubly-linked circular list. The SnapshotList
+  // implementation operates on the next/previous fields direcly.
   SnapshotImpl* prev_;
   SnapshotImpl* next_;
 
-  SnapshotList* list_;                 // just for sanity checks
+  const SequenceNumber sequence_number_;
+
+#if !defined(NDEBUG)
+  SnapshotList* list_ = nullptr;
+#endif  // !defined(NDEBUG)
 };
 
 class SnapshotList {
  public:
-  SnapshotList() {
-    list_.prev_ = &list_;
-    list_.next_ = &list_;
+  SnapshotList() : head_(0) {
+    head_.prev_ = &head_;
+    head_.next_ = &head_;
   }
 
-  bool empty() const { return list_.next_ == &list_; }
-  SnapshotImpl* oldest() const { assert(!empty()); return list_.next_; }
-  SnapshotImpl* newest() const { assert(!empty()); return list_.prev_; }
-
-  const SnapshotImpl* New(SequenceNumber seq) {
-    SnapshotImpl* s = new SnapshotImpl;
-    s->number_ = seq;
-    s->list_ = this;
-    s->next_ = &list_;
-    s->prev_ = list_.prev_;
-    s->prev_->next_ = s;
-    s->next_->prev_ = s;
-    return s;
+  bool empty() const { return head_.next_ == &head_; }
+  SnapshotImpl* oldest() const { assert(!empty()); return head_.next_; }
+  SnapshotImpl* newest() const { assert(!empty()); return head_.prev_; }
+
+  // Creates a SnapshotImpl and appends it to the end of the list.
+  SnapshotImpl* New(SequenceNumber sequence_number) {
+    assert(empty() || newest()->sequence_number_ <= sequence_number);
+
+    SnapshotImpl* snapshot = new SnapshotImpl(sequence_number);
+
+#if !defined(NDEBUG)
+    snapshot->list_ = this;
+#endif  // !defined(NDEBUG)
+    snapshot->next_ = &head_;
+    snapshot->prev_ = head_.prev_;
+    snapshot->prev_->next_ = snapshot;
+    snapshot->next_->prev_ = snapshot;
+    return snapshot;
   }
 
-  void Delete(const SnapshotImpl* s) {
-    assert(s->list_ == this);
-    s->prev_->next_ = s->next_;
-    s->next_->prev_ = s->prev_;
-    delete s;
+  // Removes a SnapshotImpl from this list.
+  //
+  // The snapshot must have been created by calling New() on this list.
+  //
+  // The snapshot pointer should not be const, because its memory is
+  // deallocated. However, that would force us to change DB::ReleaseSnapshot(),
+  // which is in the API, and currently takes a const Snapshot.
+  void Delete(const SnapshotImpl* snapshot) {
+#if !defined(NDEBUG)
+    assert(snapshot->list_ == this);
+#endif  // !defined(NDEBUG)
+    snapshot->prev_->next_ = snapshot->next_;
+    snapshot->next_->prev_ = snapshot->prev_;
+    delete snapshot;
   }
 
  private:
   // Dummy head of doubly-linked list of snapshots
-  SnapshotImpl list_;
+  SnapshotImpl head_;
 };
 
 }  // namespace leveldb