kv: fix iterator invalidation in memdb.

Signed-off-by: Haodong Tang <haodong.tang@intel.com>

diff --git a/src/kv/MemDB.cc b/src/kv/MemDB.cc
index b6a8165f9c25aeb8eb7cf87f4e3fd150158592f5..09830ce5c82c4a82ac4063e5c805c632162e875e 100644 (file)
--- a/src/kv/MemDB.cc
+++ b/src/kv/MemDB.cc
@@ -270,7 +270,7 @@ int MemDB::_setkey(ms_op_t &op)
   }
 
   m_btree[key] = bufferptr((char *) bl.c_str(), bl.length());
-
+  iterator_seq_no++;
   return 0;
 }
 
@@ -284,6 +284,7 @@ int MemDB::_rmkey(ms_op_t &op)
     assert(m_total_bytes >= bl_old.length());
     m_total_bytes -= bl_old.length();
   }
+  iterator_seq_no++;
   /*
    * Erase will call the destructor for bufferptr.
    */
@@ -341,6 +342,7 @@ int MemDB::_merge(ms_op_t &op)
 
   assert((int64_t)m_total_bytes + bytes_adjusted >= 0);
   m_total_bytes += bytes_adjusted;
+  iterator_seq_no++;
   return 0;
 }
 
@@ -403,6 +405,18 @@ bool MemDB::MDBWholeSpaceIteratorImpl::valid()
   return true;
 }
 
+bool MemDB::MDBWholeSpaceIteratorImpl::iterator_validate() {
+  if (this_seq_no != *global_seq_no) {
+    auto key = m_key_value.first;
+    assert(!key.empty());
+    m_iter = m_btree_p->lower_bound(key);
+    if (m_iter == m_btree_p->end()) {
+      return false;
+    }
+  }
+  return true;
+}
+
 void
 MemDB::MDBWholeSpaceIteratorImpl::free_last()
 {
@@ -442,6 +456,10 @@ bufferlist MemDB::MDBWholeSpaceIteratorImpl::value()
 int MemDB::MDBWholeSpaceIteratorImpl::next()
 {
   std::lock_guard<std::mutex> l(*m_btree_lock_p);
+  if (!iterator_validate()) {
+    free_last();
+    return -1;
+  }
   free_last();
   m_iter++;
   if (m_iter != m_btree_p->end()) {
@@ -455,6 +473,10 @@ int MemDB::MDBWholeSpaceIteratorImpl::next()
 int MemDB::MDBWholeSpaceIteratorImpl:: prev()
 {
   std::lock_guard<std::mutex> l(*m_btree_lock_p);
+  if (!iterator_validate()) {
+    free_last();
+    return -1;
+  }
   free_last();
   m_iter--;
   if (m_iter != m_btree_p->end()) {
@@ -477,7 +499,7 @@ int MemDB::MDBWholeSpaceIteratorImpl::seek_to_first(const std::string &k)
   } else {
     m_iter = m_btree_p->lower_bound(k);
   }
-
+  
   if (m_iter == m_btree_p->end()) {
     return -1;
   }
@@ -487,7 +509,6 @@ int MemDB::MDBWholeSpaceIteratorImpl::seek_to_first(const std::string &k)
 int MemDB::MDBWholeSpaceIteratorImpl::seek_to_last(const std::string &k)
 {
   std::lock_guard<std::mutex> l(*m_btree_lock_p);
-
   free_last();
   if (k.empty()) {
     m_iter = m_btree_p->end();

diff --git a/src/kv/MemDB.h b/src/kv/MemDB.h
index ee276431fc2403377ac4afbac54455d8352c61a4..b17c16e574ca4c6d4635e909bd4e74617c73a121 100644 (file)
--- a/src/kv/MemDB.h
+++ b/src/kv/MemDB.h
@@ -48,10 +48,11 @@ class MemDB : public KeyValueDB
   void _encode(btree::btree_map<string, bufferptr>:: iterator iter, bufferlist &bl);
   void _save();
   int _load();
+  uint64_t iterator_seq_no;
 
 public:
   MemDB(CephContext *c, const string &path, void *p) :
-    m_cct(c), m_priv(p), m_db_path(path)
+    m_cct(c), m_priv(p), m_db_path(path), iterator_seq_no(1)
   {
     //Nothing as of now
   }
@@ -132,12 +133,17 @@ public:
       std::pair<string, bufferlist> m_key_value;
       btree::btree_map<std::string, bufferptr> *m_btree_p;
       std::mutex *m_btree_lock_p;
+      uint64_t *global_seq_no;
+      uint64_t this_seq_no;
 
   public:
     MDBWholeSpaceIteratorImpl(btree::btree_map<std::string, bufferptr> *btree_p,
-                             std::mutex *btree_lock_p) {
+                             std::mutex *btree_lock_p, uint64_t *iterator_seq_no) {
         m_btree_p = btree_p;
         m_btree_lock_p = btree_lock_p;
+       std::lock_guard<std::mutex> l(*m_btree_lock_p);
+       global_seq_no = iterator_seq_no;
+       this_seq_no = *iterator_seq_no;
     }
 
     void fill_current();
@@ -147,12 +153,13 @@ public:
     int seek_to_first(const std::string &k);
     int seek_to_last(const std::string &k);
 
-    int seek_to_first() { return seek_to_first(NULL); };
-    int seek_to_last() { return seek_to_last(NULL); };
+    int seek_to_first() { return seek_to_first(std::string()); };
+    int seek_to_last() { return seek_to_last(std::string()); };
 
     int upper_bound(const std::string &prefix, const std::string &after);
     int lower_bound(const std::string &prefix, const std::string &to);
     bool valid();
+    bool iterator_validate();
 
     int next();
     int prev();
@@ -183,7 +190,7 @@ protected:
 
   WholeSpaceIterator _get_iterator() {
     return std::shared_ptr<KeyValueDB::WholeSpaceIteratorImpl>(
-      new MDBWholeSpaceIteratorImpl(&m_btree, &m_lock));
+      new MDBWholeSpaceIteratorImpl(&m_btree, &m_lock, &iterator_seq_no));
   }
 
   WholeSpaceIterator _get_snapshot_iterator();