os: KeyValueDB: allow finer-grained control of transaction operations

This patch introduces the possibility of using single key/value
modification operations into the transaction interface.

Until now, any 'set' or 'rmkeys' operations required a map of keys to be
provided to the function, which made the task of removing or setting a
bunch of keys easier. Doing these same operations for a single key,
however, would entail creating a map with a single key.

Instead, this patch adds two new virtual abstract functions, to be
implemented by derivative classes, which set or remove one single
key/value, and we then implement the map-based, existing functions in
terms of these new functions.

We also update the derivative classes of KeyValueDB in order to reflect
these changes (i.e., LevelDBStore and KeyValueDBMemory).

Signed-off-by: Joao Eduardo Luis <joao.luis@inktank.com>

src/os/KeyValueDB.h

@@ -1,4 +1,5 @@
 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
 #ifndef KEY_VALUE_DB_H
 #define KEY_VALUE_DB_H
 
@@ -20,15 +21,37 @@ public:
   class TransactionImpl {
   public:
     /// Set Keys
-    virtual void set(
+    void set(
       const string &prefix,                 ///< [in] Prefix for keys
       const std::map<string, bufferlist> &to_set ///< [in] keys/values to set
+    ) {
+      std::map<string, bufferlist>::const_iterator it;
+      for (it = to_set.begin(); it != to_set.end(); ++it)
+	set(prefix, it->first, it->second);
+    }
+
+    /// Set Key
+    virtual void set(
+      const string &prefix,   ///< [in] Prefix for the key
+      const string &k,	      ///< [in] Key to set
+      const bufferlist &bl    ///< [in] Value to set
       ) = 0;
 
+
     /// Removes Keys
-    virtual void rmkeys(
+    void rmkeys(
       const string &prefix,   ///< [in] Prefix to search for
       const std::set<string> &keys ///< [in] Keys to remove
+    ) {
+      std::set<string>::const_iterator it;
+      for (it = keys.begin(); it != keys.end(); ++it)
+	rmkey(prefix, *it);
+    }
+
+    /// Remove Key
+    virtual void rmkey(
+      const string &prefix,   ///< [in] Prefix to search for
+      const string &k	      ///< [in] Key to remove
       ) = 0;
 
     /// Removes keys beginning with prefix

src/os/LevelDBStore.cc

@@ -1,5 +1,5 @@
 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-
+// vim: ts=8 sw=2 smarttab
 #include "LevelDBStore.h"
 
 #include <set>
@@ -28,31 +28,25 @@ int LevelDBStore::init(ostream &out)
 
 void LevelDBStore::LevelDBTransactionImpl::set(
   const string &prefix,
-  const std::map<string, bufferlist> &to_set)
+  const string &k,
+  const bufferlist &to_set_bl)
 {
-  for (std::map<string, bufferlist>::const_iterator i = to_set.begin();
-       i != to_set.end();
-       ++i) {
-    buffers.push_back(i->second);
-    buffers.rbegin()->rebuild();
-    bufferlist &bl = *(buffers.rbegin());
-    string key = combine_strings(prefix, i->first);
-    keys.push_back(key);
-    bat.Delete(leveldb::Slice(*(keys.rbegin())));
-    bat.Put(leveldb::Slice(*(keys.rbegin())),
-	    leveldb::Slice(bl.c_str(), bl.length()));
-  }
+  buffers.push_back(to_set_bl);
+  buffers.rbegin()->rebuild();
+  bufferlist &bl = *(buffers.rbegin());
+  string key = combine_strings(prefix, k);
+  keys.push_back(key);
+  bat.Delete(leveldb::Slice(*(keys.rbegin())));
+  bat.Put(leveldb::Slice(*(keys.rbegin())),
+	  leveldb::Slice(bl.c_str(), bl.length()));
 }
-void LevelDBStore::LevelDBTransactionImpl::rmkeys(const string &prefix,
-						  const std::set<string> &to_rm)
+
+void LevelDBStore::LevelDBTransactionImpl::rmkey(const string &prefix,
+					         const string &k)
 {
-  for (std::set<string>::const_iterator i = to_rm.begin();
-       i != to_rm.end();
-       ++i) {
-    string key = combine_strings(prefix, *i);
-    keys.push_back(key);
-    bat.Delete(leveldb::Slice(*(keys.rbegin())));
-  }
+  string key = combine_strings(prefix, k);
+  keys.push_back(key);
+  bat.Delete(leveldb::Slice(*(keys.rbegin())));
 }
 
 void LevelDBStore::LevelDBTransactionImpl::rmkeys_by_prefix(const string &prefix)

src/os/LevelDBStore.h

@@ -36,12 +36,11 @@ public:
     LevelDBTransactionImpl(LevelDBStore *db) : db(db) {}
     void set(
       const string &prefix,
-      const std::map<string, bufferlist> &to_set
-      );
-    void rmkeys(
+      const string &k,
+      const bufferlist &bl);
+    void rmkey(
       const string &prefix,
-      const std::set<string> &keys
-      );
+      const string &k);
     void rmkeys_by_prefix(
       const string &prefix
       );

src/test/ObjectMap/KeyValueDBMemory.cc

@@ -1,4 +1,5 @@
 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
 #include "include/encoding.h"
 #include "KeyValueDBMemory.h"
 #include <map>
@@ -133,25 +134,18 @@ int KeyValueDBMemory::get_keys(const string &prefix,
 }
 
 int KeyValueDBMemory::set(const string &prefix,
-			  const map<string, bufferlist> &to_set) {
-  for (map<string, bufferlist>::const_iterator i = to_set.begin();
-       i != to_set.end();
-       ++i) {
-    bufferlist bl = i->second;
-    db[prefix][i->first] = i->second;
-  }
+			  const string &key,
+			  const bufferlist &bl) {
+  db[prefix][key] = bl;
   return 0;
 }
 
-int KeyValueDBMemory::rmkeys(const string &prefix,
-			     const std::set<string> &keys) {
-  if (!db.count(prefix))
-    return 0;
-  for (std::set<string>::const_iterator i = keys.begin();
-       i != keys.end();
-       ++i) {
-    db[prefix].erase(*i);
-  }
+int KeyValueDBMemory::rmkey(const string &prefix,
+			    const string &key) {
+  db[prefix].erase(key);
+  if (db[prefix].size() == 0)
+    db.erase(prefix);
+
   return 0;
 }
 

src/test/ObjectMap/KeyValueDBMemory.h

@@ -1,4 +1,5 @@
 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
 #include <map>
 #include <set>
 #include <string>
@@ -28,12 +29,13 @@ public:
 
   int set(
     const string &prefix,
-    const std::map<string, bufferlist> &to_set
+    const string &key,
+    const bufferlist &bl
     );
 
-  int rmkeys(
+  int rmkey(
     const string &prefix,
-    const std::set<string> &keys
+    const string &key
     );
 
   int rmkeys_by_prefix(
@@ -51,33 +53,37 @@ public:
     struct SetOp : public Context {
       KeyValueDBMemory *db;
       string prefix;
-      std::map<string, bufferlist> to_set;
+      string key;
+      bufferlist value;
       SetOp(KeyValueDBMemory *db,
 	    const string &prefix,
-	    const std::map<string, bufferlist> &to_set)
-	: db(db), prefix(prefix), to_set(to_set) {}
+	    const string &key,
+	    const bufferlist &value)
+	: db(db), prefix(prefix), key(key), value(value) {}
       void finish(int r) {
-	db->set(prefix, to_set);
+	db->set(prefix, key, value);
       }
     };
-    void set(const string &prefix, const std::map<string, bufferlist> &to_set) {
-      on_commit.push_back(new SetOp(db, prefix, to_set));
+
+    void set(const string &prefix, const string &k, const bufferlist& bl) {
+      on_commit.push_back(new SetOp(db, prefix, k, bl));
     }
 
     struct RmKeysOp : public Context {
       KeyValueDBMemory *db;
       string prefix;
-      std::set<string> keys;
+      string key;
       RmKeysOp(KeyValueDBMemory *db,
 	       const string &prefix,
-	       const std::set<string> &keys)
-	: db(db), prefix(prefix), keys(keys) {}
+	       const string &key)
+	: db(db), prefix(prefix), key(key) {}
       void finish(int r) {
-	db->rmkeys(prefix, keys);
+	db->rmkey(prefix, key);
       }
     };
-    void rmkeys(const string &prefix, const std::set<string> &to_remove) {
-      on_commit.push_back(new RmKeysOp(db, prefix, to_remove));
+
+    void rmkey(const string &prefix, const string &key) {
+      on_commit.push_back(new RmKeysOp(db, prefix, key));
     }
 
     struct RmKeysByPrefixOp : public Context {