Fix a bug that Prev() can hang.

Summary: Prev() now can hang when there is a key with more than max_skipped number of appearance internally but all of them are newer than the sequence ID to seek. Add unit tests to confirm the bug and fix it.

Test Plan: make all check

Reviewers: igor, haobo

Reviewed By: igor

CC: ljin, yhchiang, leveldb

Differential Revision: https://reviews.facebook.net/D16899

Author: siying

db/db_iter.cc

@@ -379,10 +379,10 @@ void DBIter::FindPrevUserEntry() {
   uint64_t num_skipped = 0;
 
   ValueType value_type = kTypeDeletion;
+  bool saved_key_valid = true;
   if (iter_->Valid()) {
     do {
       ParsedInternalKey ikey;
-      bool saved_key_cleared = false;
       if (ParseKey(&ikey) && ikey.sequence <= sequence_) {
         if ((value_type != kTypeDeletion) &&
             user_comparator_->Compare(ikey.user_key, saved_key_) < 0) {
@@ -393,7 +393,7 @@ void DBIter::FindPrevUserEntry() {
         if (value_type == kTypeDeletion) {
           saved_key_.clear();
           ClearSavedValue();
-          saved_key_cleared = true;
+          saved_key_valid = false;
         } else {
           Slice raw_value = iter_->value();
           if (saved_value_.capacity() > raw_value.size() + 1048576) {
@@ -403,13 +403,17 @@ void DBIter::FindPrevUserEntry() {
           SaveKey(ExtractUserKey(iter_->key()), &saved_key_);
           saved_value_.assign(raw_value.data(), raw_value.size());
         }
+      } else {
+        // In the case of ikey.sequence > sequence_, we might have already
+        // iterated to a different user key.
+        saved_key_valid = false;
       }
       num_skipped++;
       // If we have sequentially iterated via numerous keys and still not
       // found the prev user-key, then it is better to seek so that we can
       // avoid too many key comparisons. We seek to the first occurence of
       // our current key by looking for max sequence number.
-      if (!saved_key_cleared && num_skipped > max_skip_) {
+      if (saved_key_valid && num_skipped > max_skip_) {
         num_skipped = 0;
         std::string last_key;
         AppendInternalKey(&last_key,

db/db_test.cc

@@ -1368,6 +1368,75 @@ TEST(DBTest, IterSeekBeforePrev) {
   delete iter;
 }
 
+TEST(DBTest, IterNextWithNewerSeq) {
+  ASSERT_OK(Put("0", "0"));
+  dbfull()->Flush(FlushOptions());
+  ASSERT_OK(Put("a", "b"));
+  ASSERT_OK(Put("c", "d"));
+  ASSERT_OK(Put("d", "e"));
+  auto iter = db_->NewIterator(ReadOptions());
+
+  // Create a key that needs to be skipped for Seq too new
+  for (uint64_t i = 0; i < last_options_.max_sequential_skip_in_iterations + 1;
+       i++) {
+    ASSERT_OK(Put("b", "f"));
+  }
+
+  iter->Seek(Slice("a"));
+  ASSERT_EQ(IterStatus(iter), "a->b");
+  iter->Next();
+  ASSERT_EQ(IterStatus(iter), "c->d");
+  delete iter;
+}
+
+TEST(DBTest, IterPrevWithNewerSeq) {
+  ASSERT_OK(Put("0", "0"));
+  dbfull()->Flush(FlushOptions());
+  ASSERT_OK(Put("a", "b"));
+  ASSERT_OK(Put("c", "d"));
+  ASSERT_OK(Put("d", "e"));
+  auto iter = db_->NewIterator(ReadOptions());
+
+  // Create a key that needs to be skipped for Seq too new
+  for (uint64_t i = 0; i < last_options_.max_sequential_skip_in_iterations + 1;
+       i++) {
+    ASSERT_OK(Put("b", "f"));
+  }
+
+  iter->Seek(Slice("d"));
+  ASSERT_EQ(IterStatus(iter), "d->e");
+  iter->Prev();
+  ASSERT_EQ(IterStatus(iter), "c->d");
+  iter->Prev();
+  ASSERT_EQ(IterStatus(iter), "a->b");
+
+  iter->Prev();
+  delete iter;
+}
+
+TEST(DBTest, IterPrevWithNewerSeq2) {
+  ASSERT_OK(Put("0", "0"));
+  dbfull()->Flush(FlushOptions());
+  ASSERT_OK(Put("a", "b"));
+  ASSERT_OK(Put("c", "d"));
+  ASSERT_OK(Put("d", "e"));
+  auto iter = db_->NewIterator(ReadOptions());
+  iter->Seek(Slice("c"));
+  ASSERT_EQ(IterStatus(iter), "c->d");
+
+  // Create a key that needs to be skipped for Seq too new
+  for (uint64_t i = 0; i < last_options_.max_sequential_skip_in_iterations + 1;
+      i++) {
+    ASSERT_OK(Put("b", "f"));
+  }
+
+  iter->Prev();
+  ASSERT_EQ(IterStatus(iter), "a->b");
+
+  iter->Prev();
+  delete iter;
+}
+
 TEST(DBTest, IterEmpty) {
   do {
     Iterator* iter = db_->NewIterator(ReadOptions());