Merge pull request #989 from ceph/wip-7056

osd/ReplicatedPG: include omap header in copy-get

This now passes rados/thrash tests without failures.

src/osd/ReplicatedPG.cc

@@ -4676,11 +4676,14 @@ int ReplicatedPG::fill_in_copy_get(bufferlist::iterator& bp, OSDOp& osd_op,
   std::map<std::string,bufferlist>& out_omap = reply_obj.omap;
   if (left > 0 && !cursor.omap_complete) {
     assert(cursor.data_complete);
+    if (cursor.omap_offset.empty()) {
+      osd->store->omap_get_header(coll, oi.soid, &reply_obj.omap_header);
+    }
     ObjectMap::ObjectMapIterator iter =
       osd->store->get_omap_iterator(coll, oi.soid);
     assert(iter);
+    iter->upper_bound(cursor.omap_offset);
     if (iter->valid()) {
-      iter->upper_bound(cursor.omap_offset);
       for (; left > 0 && iter->valid(); iter->next()) {
 	out_omap.insert(make_pair(iter->key(), iter->value()));
 	left -= iter->key().length() + 4 + iter->value().length() + 4;
@@ -4695,12 +4698,17 @@ int ReplicatedPG::fill_in_copy_get(bufferlist::iterator& bp, OSDOp& osd_op,
   }
 
   dout(20) << " cursor.is_complete=" << cursor.is_complete()
-		     << " " << out_attrs.size() << " attrs"
-		     << " " << bl.length() << " bytes"
-		     << " " << out_omap.size() << " keys"
-		     << dendl;
+	   << " " << out_attrs.size() << " attrs"
+	   << " " << bl.length() << " bytes"
+	   << " " << reply_obj.omap_header.length() << " omap header bytes"
+	   << " " << out_omap.size() << " keys"
+	   << dendl;
   reply_obj.cursor = cursor;
   if (classic) {
+    if (reply_obj.omap_header.length() > 0) {
+      derr << oi.soid << " omap header being dropped by classic copy-get api"
+	   << dendl;
+    }
     reply_obj.encode_classic(osd_op.outdata);
   } else {
     ::encode(reply_obj, osd_op.outdata);
@@ -4749,7 +4757,7 @@ void ReplicatedPG::_copy_some(ObjectContextRef obc, CopyOpRef cop)
   op.copy_get(&cop->cursor, cct->_conf->osd_copyfrom_max_chunk,
 	      &cop->results->object_size, &cop->results->mtime,
 	      &cop->results->category,
-	      &cop->attrs, &cop->data, &cop->omap,
+	      &cop->attrs, &cop->data, &cop->omap_header, &cop->omap,
 	      &cop->rval);
 
   C_Copyfrom *fin = new C_Copyfrom(this, obc->obs.oi.soid,
@@ -4842,6 +4850,10 @@ void ReplicatedPG::_write_copy_chunk(CopyOpRef cop, ObjectStore::Transaction *t)
     cop->data.clear();
   }
   if (!cop->temp_cursor.omap_complete) {
+    if (cop->omap_header.length()) {
+      t->omap_setheader(cop->temp_coll, cop->temp_oid, cop->omap_header);
+      cop->omap_header.clear();
+    }
     t->omap_setkeys(cop->temp_coll, cop->temp_oid, cop->omap);
     cop->omap.clear();
   }

src/osd/ReplicatedPG.h

@@ -133,6 +133,7 @@ public:
     object_copy_cursor_t cursor;
     map<string,bufferlist> attrs;
     bufferlist data;
+    bufferlist omap_header;
     map<string,bufferlist> omap;
     int rval;
 

src/osd/osd_types.cc

@@ -2676,7 +2676,7 @@ void object_copy_data_t::decode_classic(bufferlist::iterator& bl)
 
 void object_copy_data_t::encode(bufferlist& bl) const
 {
-  ENCODE_START(1, 1, bl);
+  ENCODE_START(2, 1, bl);
   ::encode(size, bl);
   ::encode(mtime, bl);
   ::encode(category, bl);
@@ -2684,12 +2684,13 @@ void object_copy_data_t::encode(bufferlist& bl) const
   ::encode(data, bl);
   ::encode(omap, bl);
   ::encode(cursor, bl);
+  ::encode(omap_header, bl);
   ENCODE_FINISH(bl);
 }
 
 void object_copy_data_t::decode(bufferlist::iterator& bl)
 {
-  DECODE_START(1, bl);
+  DECODE_START(2, bl);
   ::decode(size, bl);
   ::decode(mtime, bl);
   ::decode(category, bl);
@@ -2697,6 +2698,8 @@ void object_copy_data_t::decode(bufferlist::iterator& bl)
   ::decode(data, bl);
   ::decode(omap, bl);
   ::decode(cursor, bl);
+  if (struct_v >= 2)
+    ::decode(omap_header, bl);
   DECODE_FINISH(bl);
 }
 
@@ -2725,6 +2728,7 @@ void object_copy_data_t::generate_test_instances(list<object_copy_data_t*>& o)
   o.back()->omap["why"] = bl2;
   bufferptr databp("iamsomedatatocontain", 20);
   o.back()->data.push_back(databp);
+  o.back()->omap_header.append("this is an omap header");
 }
 
 void object_copy_data_t::dump(Formatter *f) const
@@ -2738,6 +2742,7 @@ void object_copy_data_t::dump(Formatter *f) const
      const-correctness prents that */
   f->dump_int("attrs_size", attrs.size());
   f->dump_int("omap_size", omap.size());
+  f->dump_int("omap_header_length", omap_header.length());
   f->dump_int("data_length", data.length());
 }
 

src/osd/osd_types.h

@@ -1953,6 +1953,7 @@ struct object_copy_data_t {
   utime_t mtime;
   map<string, bufferlist> attrs;
   bufferlist data;
+  bufferlist omap_header;
   map<string, bufferlist> omap;
   string category;
 public:

src/osdc/Objecter.h

@@ -571,7 +571,7 @@ struct ObjectOperation {
     utime_t *out_mtime;
     string *out_category;
     std::map<std::string,bufferlist> *out_attrs;
-    bufferlist *out_data;
+    bufferlist *out_data, *out_omap_header;
     std::map<std::string,bufferlist> *out_omap;
     int *prval;
     C_ObjectOperation_copyget(object_copy_cursor_t *c,
@@ -579,12 +579,13 @@ struct ObjectOperation {
 			      utime_t *m,
 			      string *cat,
 			      std::map<std::string,bufferlist> *a,
-			      bufferlist *d,
+			      bufferlist *d, bufferlist *oh,
 			      std::map<std::string,bufferlist> *o,
 			      int *r)
       : cursor(c),
 	out_size(s), out_mtime(m), out_category(cat),
-	out_attrs(a), out_data(d), out_omap(o), prval(r) {}
+	out_attrs(a), out_data(d), out_omap_header(oh),
+	out_omap(o), prval(r) {}
     void finish(int r) {
       if (r < 0)
 	return;
@@ -602,6 +603,8 @@ struct ObjectOperation {
 	  *out_attrs = copy_reply.attrs;
 	if (out_data)
 	  out_data->claim_append(copy_reply.data);
+	if (out_omap_header)
+	  out_omap_header->claim_append(copy_reply.omap_header);
 	if (out_omap)
 	  *out_omap = copy_reply.omap;
 	*cursor = copy_reply.cursor;
@@ -619,6 +622,7 @@ struct ObjectOperation {
 		string *out_category,
 		std::map<std::string,bufferlist> *out_attrs,
 		bufferlist *out_data,
+		bufferlist *out_omap_header,
 		std::map<std::string,bufferlist> *out_omap,
 		int *prval) {
     OSDOp& osd_op = add_op(CEPH_OSD_OP_COPY_GET);
@@ -629,7 +633,8 @@ struct ObjectOperation {
     out_rval[p] = prval;
     C_ObjectOperation_copyget *h =
       new C_ObjectOperation_copyget(cursor, out_size, out_mtime, out_category,
-                                    out_attrs, out_data, out_omap, prval);
+                                    out_attrs, out_data, out_omap_header,
+				    out_omap, prval);
     out_bl[p] = &h->bl;
     out_handler[p] = h;
   }

src/test/osd/RadosModel.h

@@ -1119,12 +1119,12 @@ public:
 	assert(old_value.header == header);
       }
       if (omap.size() != old_value.attrs.size()) {
-	cerr << num << ": oid " << oid << " tmap.size() is " << omap.size()
+	cerr << num << ": oid " << oid << " omap.size() is " << omap.size()
 	     << " and old is " << old_value.attrs.size() << std::endl;
 	assert(omap.size() == old_value.attrs.size());
       }
       if (omap_keys.size() != old_value.attrs.size()) {
-	cerr << num << ": oid " << oid << " tmap.size() is " << omap_keys.size()
+	cerr << num << ": oid " << oid << " omap.size() is " << omap_keys.size()
 	     << " and old is " << old_value.attrs.size() << std::endl;
 	assert(omap_keys.size() == old_value.attrs.size());
       }