mon/MonitorDBStore: allow compaction of ranges

Allow a transaction to describe the compaction of a range of keys.  Do this
in a backward compatible say, such that older code will interpret the
compaction of a prefix + range as compaction of the entire prefix.  This
allows us to avoid introducing any new feature bits.

Signed-off-by: Sage Weil <sage@inktank.com>

diff --git a/src/mon/MonitorDBStore.h b/src/mon/MonitorDBStore.h
index 3ed144aa016029484393a41d0e79aa1e54e17104..ed8be193ab64ecf407eba51c710b8dc2cdd72d32 100644 (file)
--- a/src/mon/MonitorDBStore.h
+++ b/src/mon/MonitorDBStore.h
@@ -38,7 +38,7 @@ class MonitorDBStore
   struct Op {
     uint8_t type;
     string prefix;
-    string key;
+    string key, endkey;
     bufferlist bl;
 
     Op() { }
@@ -46,22 +46,27 @@ class MonitorDBStore
       : type(t), prefix(p), key(k) { }
     Op(int t, const string& p, string k, bufferlist& b)
       : type(t), prefix(p), key(k), bl(b) { }
+    Op(int t, const string& p, string start, string end)
+      : type(t), prefix(p), key(start), endkey(end) { }
 
     void encode(bufferlist& encode_bl) const {
-      ENCODE_START(1, 1, encode_bl);
+      ENCODE_START(2, 1, encode_bl);
       ::encode(type, encode_bl);
       ::encode(prefix, encode_bl);
       ::encode(key, encode_bl);
       ::encode(bl, encode_bl);
+      ::encode(endkey, encode_bl);
       ENCODE_FINISH(encode_bl);
     }
 
     void decode(bufferlist::iterator& decode_bl) {
-      DECODE_START(1, decode_bl);
+      DECODE_START(2, decode_bl);
       ::decode(type, decode_bl);
       ::decode(prefix, decode_bl);
       ::decode(key, decode_bl);
       ::decode(bl, decode_bl);
+      if (struct_v >= 2)
+       ::decode(endkey, decode_bl);
       DECODE_FINISH(decode_bl);
     }
   };
@@ -72,7 +77,7 @@ class MonitorDBStore
     enum {
       OP_PUT   = 1,
       OP_ERASE = 2,
-      OP_COMPACT_PREFIX = 3,
+      OP_COMPACT = 3,
     };
 
     void put(string prefix, string key, bufferlist& bl) {
@@ -102,7 +107,11 @@ class MonitorDBStore
     }
 
     void compact_prefix(string prefix) {
-      ops.push_back(Op(OP_COMPACT_PREFIX, prefix, string()));
+      ops.push_back(Op(OP_COMPACT, prefix, string()));
+    }
+
+    void compact_range(string prefix, string start, string end) {
+      ops.push_back(Op(OP_COMPACT, prefix, start, end));
     }
 
     void encode(bufferlist& bl) const {
@@ -166,10 +175,12 @@ class MonitorDBStore
            f->dump_string("key", op.key);
          }
          break;
-       case OP_COMPACT_PREFIX:
+       case OP_COMPACT:
          {
-           f->dump_string("type", "COMPACT_PREFIX");
+           f->dump_string("type", "COMPACT");
            f->dump_string("prefix", op.prefix);
+           f->dump_string("start", op.key);
+           f->dump_string("end", op.endkey);
          }
          break;
        default:
@@ -195,7 +206,7 @@ class MonitorDBStore
       bl.write_fd(dump_fd);
     }
 
-    list<string> compact_prefixes;
+    list<pair<string, pair<string,string> > > compact;
     for (list<Op>::const_iterator it = t.ops.begin(); it != t.ops.end(); ++it) {
       const Op& op = *it;
       switch (op.type) {
@@ -205,8 +216,8 @@ class MonitorDBStore
       case Transaction::OP_ERASE:
        dbt->rmkey(op.prefix, op.key);
        break;
-      case Transaction::OP_COMPACT_PREFIX:
-       compact_prefixes.push_back(op.prefix);
+      case Transaction::OP_COMPACT:
+       compact.push_back(make_pair(op.prefix, make_pair(op.key, op.endkey)));
        break;
       default:
        derr << __func__ << " unknown op type " << op.type << dendl;
@@ -216,9 +227,13 @@ class MonitorDBStore
     }
     int r = db->submit_transaction_sync(dbt);
     if (r >= 0) {
-      while (!compact_prefixes.empty()) {
-       db->compact_prefix_async(compact_prefixes.front());
-       compact_prefixes.pop_front();
+      while (!compact.empty()) {
+       if (compact.front().second.first == string() &&
+           compact.front().second.second == string())
+         db->compact_prefix_async(compact.front().first);
+       else
+         db->compact_range_async(compact.front().first, compact.front().second.first, compact.front().second.second);
+       compact.pop_front();
       }
     }
     return r;