osd: mux/demux OSDOp::outdata in MOSDOpReply

Bump encoding, so that we don't try to demux old encoded messages, which
will likely have OSDOp::payload_len == indata.length().

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

diff --git a/src/messages/MOSDOpReply.h b/src/messages/MOSDOpReply.h
index e8a0930c897f3ffe12afdcab8f638c9d5dbc4d9f..b179a2cd8ef8da9b9e5a99c5ba26164fa994befa 100644 (file)
--- a/src/messages/MOSDOpReply.h
+++ b/src/messages/MOSDOpReply.h
@@ -58,6 +58,16 @@ public:
 
   void add_flags(int f) { flags |= f; }
 
+  void claim_op_out_data(vector<OSDOp>& o) {
+    assert(ops.size() == o.size());
+    for (unsigned i = 0; i < o.size(); i++) {
+      ops[i].outdata.claim(o[i].outdata);
+    }
+  }
+  void claim_ops(vector<OSDOp>& o) {
+    o.swap(ops);
+  }
+
   /**
    * get retry attempt
    *
@@ -88,6 +98,10 @@ public:
     osdmap_epoch = e;
     reassert_version = req->reassert_version;
     retry_attempt = req->get_retry_attempt();
+
+    // zero out ops payload_len
+    for (unsigned i = 0; i < ops.size(); i++)
+      ops[i].op.payload_len = 0;
   }
   MOSDOpReply() {}
 private:
@@ -95,6 +109,9 @@ private:
 
 public:
   virtual void encode_payload(CephContext *cct) {
+
+    merge_osd_op_vector_out_data(ops, data);
+
     if (!connection->has_feature(CEPH_FEATURE_PGID64)) {
       ceph_osd_reply_head head;
       memset(&head, 0, sizeof(head));
@@ -110,7 +127,7 @@ public:
       }
       ::encode_nohead(oid.name, payload);
     } else {
-      header.version = 3;
+      header.version = 4;
       ::encode(oid, payload);
       ::encode(pgid, payload);
       ::encode(flags, payload);
@@ -124,6 +141,9 @@ public:
        ::encode(ops[i].op, payload);
 
       ::encode(retry_attempt, payload);
+
+      for (unsigned i = 0; i < num_ops; i++)
+       ::encode(ops[i].rval, payload);
     }
   }
   virtual void decode_payload(CephContext *cct) {
@@ -160,6 +180,13 @@ public:
        ::decode(retry_attempt, p);
       else
        retry_attempt = -1;
+
+      if (header.version >= 4) {
+       for (unsigned i = 0; i < num_ops; ++i)
+         ::decode(ops[i].rval, p);
+
+       split_osd_op_vector_out_data(ops, data);
+      }
     }
   }