osd: add tmap key (exclusive) create

Return EEXIST if a key already exists.

Signed-off-by: Sage Weil <sage@newdream.net>

diff --git a/src/include/rados.h b/src/include/rados.h
index a2aeb8afa13518f5ca12a7074ce2b15e6e16361e..0ed075f553c3547608f9e89e2f8155808a4aa00c 100644 (file)
--- a/src/include/rados.h
+++ b/src/include/rados.h
@@ -300,6 +300,7 @@ static inline int ceph_osd_op_mode_modify(int op)
  */
 #define CEPH_OSD_TMAP_HDR 'h'
 #define CEPH_OSD_TMAP_SET 's'
+#define CEPH_OSD_TMAP_CREATE 'c' /* create key */
 #define CEPH_OSD_TMAP_RM  'r'
 
 extern const char *ceph_osd_op_name(int op);

diff --git a/src/include/rados/librados.h b/src/include/rados/librados.h
index b64ad40355cd1daec214f0184e981543326b5a19..a66d32b77d3ea09a69516606ee884298133717fd 100644 (file)
--- a/src/include/rados/librados.h
+++ b/src/include/rados/librados.h
@@ -12,6 +12,7 @@ extern "C" {
 #ifndef CEPH_OSD_TMAP_SET
 #define CEPH_OSD_TMAP_HDR 'h'
 #define CEPH_OSD_TMAP_SET 's'
+#define CEPH_OSD_TMAP_CREATE 'c'
 #define CEPH_OSD_TMAP_RM  'r'
 #endif
 

diff --git a/src/osd/ReplicatedPG.cc b/src/osd/ReplicatedPG.cc
index 0c80d4c5ed12708d898fe17e38ec422641a9009d..a6210cdc80860ebf3501b4d34a5ef002c1f92f18 100644 (file)
--- a/src/osd/ReplicatedPG.cc
+++ b/src/osd/ReplicatedPG.cc
@@ -1807,6 +1807,18 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops,
              }
              break;
              
+           case CEPH_OSD_TMAP_CREATE: // create (new) key
+             {
+               ::decode(key, bp);
+               bufferlist data;
+               ::decode(data, bp);
+               if (m.count(key))
+                 return -EEXIST;
+               m[key] = data;
+               changed = true;
+             }
+             break;
+
            case CEPH_OSD_TMAP_RM: // remove key
              ::decode(key, bp);
              if (m.count(key)) {
@@ -1870,8 +1882,8 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops,
            dout(10) << "tmapup op " << (int)op << " key " << key << dendl;
 
            // skip existing intervening keys
-           bool stop = false;
-           while (have_next && !stop) {
+           bool key_exists = false;
+           while (have_next && !key_exists) {
              dout(20) << "  (have_next=" << have_next << " nextkey=" << nextkey << ")" << dendl;
              if (nextkey > key)
                break;
@@ -1883,7 +1895,7 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops,
              } else {
                // don't copy; discard old value.  and stop.
                dout(20) << "  drop " << nextkey << " " << nextval.length() << dendl;
-               stop = true;
+               key_exists = true;
                nkeys--;
              }       
              if (!ip.end()) {
@@ -1900,6 +1912,15 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops,
              ::encode(val, newkeydata);
              dout(20) << "   set " << key << " " << val.length() << dendl;
              nkeys++;
+           } else if (op == CEPH_OSD_TMAP_CREATE) {
+             if (key_exists)
+               return -EEXIST;
+             bufferlist val;
+             ::decode(val, bp);
+             ::encode(key, newkeydata);
+             ::encode(val, newkeydata);
+             dout(20) << "   create " << key << " " << val.length() << dendl;
+             nkeys++;
            } else if (op == CEPH_OSD_TMAP_RM) {
              // do nothing.
            }