osd: verify TMAPPUT data is sorted

The MDS may try to write unsorted data; make sure it is sorted before
we write it.

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

diff --git a/src/osd/ReplicatedPG.cc b/src/osd/ReplicatedPG.cc
index 2990822a98c3d462c3ee078079014842e79fe73a..613296bff77c31282aaf849f656b9646a28d2f6a 100644 (file)
--- a/src/osd/ReplicatedPG.cc
+++ b/src/osd/ReplicatedPG.cc
@@ -2624,13 +2624,27 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops)
        //osd_op.data.hexdump(*_dout);
        //_dout_lock.Unlock();
 
-       // verify
-       if (0) {
+       // verify sort order
+       bool unsorted = false;
+       if (true) {
          bufferlist header;
-         map<string, bufferlist> m;
          ::decode(header, bp);
-         ::decode(m, bp);
-         assert(bp.end());
+         uint32_t n;
+         ::decode(n, bp);
+         string last_key;
+         while (n--) {
+           string key;
+           ::decode(key, bp);
+           dout(10) << "tmapput key " << key << dendl;
+           bufferlist val;
+           ::decode(val, bp);
+           if (key < last_key) {
+             dout(10) << "TMAPPUT is unordered; resorting" << dendl;
+             unsorted = true;
+             break;
+           }
+           last_key = key;
+         }
        }
 
        if (g_conf->osd_tmapput_sets_uses_tmap) {
@@ -2645,6 +2659,19 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops)
        newop.op.extent.offset = 0;
        newop.op.extent.length = osd_op.indata.length();
        newop.indata = osd_op.indata;
+
+       if (unsorted) {
+         bp = osd_op.indata.begin();
+         bufferlist header;
+         map<string, bufferlist> m;
+         ::decode(header, bp);
+         ::decode(m, bp);
+         assert(bp.end());
+         bufferlist newbl;
+         ::encode(header, newbl);
+         ::encode(m, newbl);
+         newop.indata = newbl;
+       }
        do_osd_ops(ctx, nops);
       }
       break;

diff --git a/src/test/librados/misc.cc b/src/test/librados/misc.cc
index 7f717bb43f2f93bf103b0b1f379726b120b73ab5..50c837014910f04eb15dde0584e776898b3fc970 100644 (file)
--- a/src/test/librados/misc.cc
+++ b/src/test/librados/misc.cc
@@ -216,6 +216,36 @@ TEST(LibRadosMisc, TmapUpdateMisorderedPP) {
   ASSERT_EQ(0, destroy_one_pool_pp(pool_name, cluster));
 }
 
+TEST(LibRadosMisc, TmapUpdateMisorderedPutPP) {
+  Rados cluster;
+  std::string pool_name = get_temp_pool_name();
+  ASSERT_EQ("", create_one_pool_pp(pool_name, cluster));
+  IoCtx ioctx;
+  cluster.ioctx_create(pool_name.c_str(), ioctx);
+
+  // create unsorted tmap
+  string h("header");
+  bufferlist bl;
+  ::encode(h, bl);
+  uint32_t n = 3;
+  ::encode(n, bl);
+  ::encode(string("b"), bl);
+  ::encode(string("bval"), bl);
+  ::encode(string("a"), bl);
+  ::encode(string("aval"), bl);
+  ::encode(string("c"), bl);
+  ::encode(string("cval"), bl);
+  ASSERT_EQ(0, ioctx.tmap_put("foo", bl));
+
+  // check
+  bufferlist newbl;
+  ASSERT_EQ(0, ioctx.read("foo", bl, 0, 0));
+  ASSERT_EQ(bl.contents_equal(newbl), false);
+
+  ioctx.close();
+  ASSERT_EQ(0, destroy_one_pool_pp(pool_name, cluster));
+}
+
 TEST(LibRadosMisc, Exec) {
   char buf[128];
   rados_t cluster;