memstore: protect object omap with a mutex

Signed-off-by: Casey Bodley <cbodley@redhat.com>

diff --git a/src/os/MemStore.cc b/src/os/MemStore.cc
index 788589d22cb1227d64b6364886d3e85f5b8204b3..d0615c536b43297b64f61abeaf61e49eb5b0da76 100644 (file)
--- a/src/os/MemStore.cc
+++ b/src/os/MemStore.cc
@@ -463,6 +463,7 @@ int MemStore::omap_get(
   ObjectRef o = c->get_object(oid);
   if (!o)
     return -ENOENT;
+  std::lock_guard<std::mutex> lock(o->omap_mutex);
   *header = o->omap_header;
   *out = o->omap;
   return 0;
@@ -484,6 +485,7 @@ int MemStore::omap_get_header(
   ObjectRef o = c->get_object(oid);
   if (!o)
     return -ENOENT;
+  std::lock_guard<std::mutex> lock(o->omap_mutex);
   *header = o->omap_header;
   return 0;
 }
@@ -503,6 +505,7 @@ int MemStore::omap_get_keys(
   ObjectRef o = c->get_object(oid);
   if (!o)
     return -ENOENT;
+  std::lock_guard<std::mutex> lock(o->omap_mutex);
   for (map<string,bufferlist>::iterator p = o->omap.begin();
        p != o->omap.end();
        ++p)
@@ -526,6 +529,7 @@ int MemStore::omap_get_values(
   ObjectRef o = c->get_object(oid);
   if (!o)
     return -ENOENT;
+  std::lock_guard<std::mutex> lock(o->omap_mutex);
   for (set<string>::const_iterator p = keys.begin();
        p != keys.end();
        ++p) {
@@ -552,6 +556,7 @@ int MemStore::omap_check_keys(
   ObjectRef o = c->get_object(oid);
   if (!o)
     return -ENOENT;
+  std::lock_guard<std::mutex> lock(o->omap_mutex);
   for (set<string>::const_iterator p = keys.begin();
        p != keys.end();
        ++p) {
@@ -1109,6 +1114,12 @@ int MemStore::_clone(coll_t cid, const ghobject_t& oldoid,
   }
   used_bytes += oo->get_size() - no->get_size();
   no->clone(oo.get(), 0, oo->get_size(), 0);
+
+  // take both omap locks with std::lock()
+  std::unique_lock<std::mutex> oo_lock(oo->omap_mutex, std::defer_lock),
+      no_lock(no->omap_mutex, std::defer_lock);
+  std::lock(oo_lock, no_lock);
+
   no->omap_header = oo->omap_header;
   no->omap = oo->omap;
   no->xattr = oo->xattr;
@@ -1160,6 +1171,7 @@ int MemStore::_omap_clear(coll_t cid, const ghobject_t &oid)
   ObjectRef o = c->get_object(oid);
   if (!o)
     return -ENOENT;
+  std::lock_guard<std::mutex> lock(o->omap_mutex);
   o->omap.clear();
   o->omap_header.clear();
   return 0;
@@ -1177,6 +1189,7 @@ int MemStore::_omap_setkeys(coll_t cid, const ghobject_t &oid,
   ObjectRef o = c->get_object(oid);
   if (!o)
     return -ENOENT;
+  std::lock_guard<std::mutex> lock(o->omap_mutex);
   for (map<string,bufferlist>::const_iterator p = aset.begin(); p != aset.end(); ++p)
     o->omap[p->first] = p->second;
   return 0;
@@ -1194,6 +1207,7 @@ int MemStore::_omap_rmkeys(coll_t cid, const ghobject_t &oid,
   ObjectRef o = c->get_object(oid);
   if (!o)
     return -ENOENT;
+  std::lock_guard<std::mutex> lock(o->omap_mutex);
   for (set<string>::const_iterator p = keys.begin(); p != keys.end(); ++p)
     o->omap.erase(*p);
   return 0;
@@ -1212,10 +1226,10 @@ int MemStore::_omap_rmkeyrange(coll_t cid, const ghobject_t &oid,
   ObjectRef o = c->get_object(oid);
   if (!o)
     return -ENOENT;
+  std::lock_guard<std::mutex> lock(o->omap_mutex);
   map<string,bufferlist>::iterator p = o->omap.lower_bound(first);
   map<string,bufferlist>::iterator e = o->omap.lower_bound(last);
-  while (p != e)
-    o->omap.erase(p++);
+  o->omap.erase(p, e);
   return 0;
 }
 
@@ -1231,6 +1245,7 @@ int MemStore::_omap_setheader(coll_t cid, const ghobject_t &oid,
   ObjectRef o = c->get_object(oid);
   if (!o)
     return -ENOENT;
+  std::lock_guard<std::mutex> lock(o->omap_mutex);
   o->omap_header = bl;
   return 0;
 }

diff --git a/src/os/MemStore.h b/src/os/MemStore.h
index 38159e9b7ce8203e524a5dc836ad327c28c8cdbd..d06f331c0687b05da21287bbe14ceb42aa8584e7 100644 (file)
--- a/src/os/MemStore.h
+++ b/src/os/MemStore.h
@@ -16,6 +16,7 @@
 #ifndef CEPH_MEMSTORE_H
 #define CEPH_MEMSTORE_H
 
+#include <mutex>
 #include <boost/intrusive_ptr.hpp>
 
 #include "include/assert.h"
@@ -29,6 +30,7 @@
 class MemStore : public ObjectStore {
 public:
   struct Object : public RefCountedObject {
+    std::mutex omap_mutex;
     map<string,bufferptr> xattr;
     bufferlist omap_header;
     map<string,bufferlist> omap;
@@ -188,35 +190,35 @@ private:
       : c(c), o(o), it(o->omap.begin()) {}
 
     int seek_to_first() {
-      RWLock::RLocker l(c->lock);
+      std::lock_guard<std::mutex>(o->omap_mutex);
       it = o->omap.begin();
       return 0;
     }
     int upper_bound(const string &after) {
-      RWLock::RLocker l(c->lock);
+      std::lock_guard<std::mutex>(o->omap_mutex);
       it = o->omap.upper_bound(after);
       return 0;
     }
     int lower_bound(const string &to) {
-      RWLock::RLocker l(c->lock);
+      std::lock_guard<std::mutex>(o->omap_mutex);
       it = o->omap.lower_bound(to);
       return 0;
     }
     bool valid() {
-      RWLock::RLocker l(c->lock);
+      std::lock_guard<std::mutex>(o->omap_mutex);
       return it != o->omap.end();      
     }
     int next() {
-      RWLock::RLocker l(c->lock);
+      std::lock_guard<std::mutex>(o->omap_mutex);
       ++it;
       return 0;
     }
     string key() {
-      RWLock::RLocker l(c->lock);
+      std::lock_guard<std::mutex>(o->omap_mutex);
       return it->first;
     }
     bufferlist value() {
-      RWLock::RLocker l(c->lock);
+      std::lock_guard<std::mutex>(o->omap_mutex);
       return it->second;
     }
     int status() {