osd: push all the read state into OpContext

Instead of referencing items on the stack, put everything important
directly in the OpContext.

diff --git a/src/osd/ReplicatedPG.cc b/src/osd/ReplicatedPG.cc
index 1785f44141ebcd91f3c20156391770589643ee3e..edf9357937a7c2bd7fa94a5557c47080c5f1711b 100644 (file)
--- a/src/osd/ReplicatedPG.cc
+++ b/src/osd/ReplicatedPG.cc
@@ -640,15 +640,14 @@ int ReplicatedPG::pick_read_snap(sobject_t& soid, object_info_t& coi)
 } 
 
 
-int ReplicatedPG::do_read_ops(MOSDOp *op, OpContext& ctx,
-                             bufferlist::iterator& bp,
-                             bufferlist& data,
-                             int *data_off)
+int ReplicatedPG::do_read_ops(ReadOpContext *ctx,
+                             bufferlist::iterator& bp, bufferlist& data)
 {
   int result = 0;
-  sobject_t& soid = ctx.soid;
-  object_info_t& oi = ctx.oi;
-  vector<ceph_osd_op> &ops = ctx.ops;
+
+  vector<ceph_osd_op>& ops = ctx->ops;
+  object_info_t& oi = ctx->oi;
+  const sobject_t& soid = oi.soid;
 
   for (vector<ceph_osd_op>::iterator p = ops.begin(); p != ops.end(); p++) {
     switch (p->op) {
@@ -657,8 +656,8 @@ int ReplicatedPG::do_read_ops(MOSDOp *op, OpContext& ctx,
        // read into a buffer
        bufferlist bl;
        int r = osd->store->read(info.pgid.to_coll(), soid, p->offset, p->length, bl);
-       if (data.length() == 0 && data_off)
-         *data_off = p->offset;
+       if (data.length() == 0)
+         ctx->data_off = p->offset;
        data.claim(bl);
        if (r >= 0) 
          p->length = r;
@@ -683,7 +682,7 @@ int ReplicatedPG::do_read_ops(MOSDOp *op, OpContext& ctx,
        //dout(20) << "rdcall param=" << indata.c_str() << dendl;
        
        ClassHandler::ClassData *cls;
-        result = osd->get_class(cname, info.pgid, op, &cls);
+        result = osd->get_class(cname, info.pgid, ctx->op, &cls);
        if (result) {
          dout(10) << "rdcall class " << cname << " does not exist" << dendl;
           if (result == -EAGAIN)
@@ -732,7 +731,7 @@ int ReplicatedPG::do_read_ops(MOSDOp *op, OpContext& ctx,
       break;
 
     case CEPH_OSD_OP_MASKTRUNC:
-      if (p != op->ops.begin()) {
+      if (p != ops.begin()) {
        ceph_osd_op& rd = *(p - 1);
        ceph_osd_op& m = *p;
        
@@ -812,19 +811,18 @@ int ReplicatedPG::do_read_ops(MOSDOp *op, OpContext& ctx,
 void ReplicatedPG::op_read(MOSDOp *op)
 {
   object_t oid = op->get_oid();
-  sobject_t soid(oid, op->get_snapid());
+  ReadOpContext ctx(op, op->ops, sobject_t(op->get_oid(), op->get_snapid()));
+  sobject_t& soid = ctx.oi.soid;
 
-  dout(10) << "op_read " << soid << " " << op->ops << dendl;
+  dout(10) << "op_read " << soid << " " << ctx.ops << dendl;
 
   bufferlist::iterator bp = op->get_data().begin();
   bufferlist data;
-  int data_off = 0;
   int result = 0;
 
   // pick revision
-  object_info_t oi(soid);
   if (soid.snap) {
-    result = pick_read_snap(soid, oi);
+    result = pick_read_snap(soid, ctx.oi);
     if (result == -EAGAIN) {
       wait_for_missing_object(soid, op);
       return;
@@ -835,7 +833,7 @@ void ReplicatedPG::op_read(MOSDOp *op)
 
   // wrlocked?
   if ((op->get_snapid() == 0 || op->get_snapid() == CEPH_NOSNAP) &&
-      block_if_wrlocked(op, oi)) 
+      block_if_wrlocked(op, ctx.oi)) 
     return;
 
 
@@ -885,20 +883,16 @@ void ReplicatedPG::op_read(MOSDOp *op)
     }
   }
 
-  {
-    OpContext ctx(soid, oi, op->ops);
-
-    // do it.
-    result = do_read_ops(op, ctx, bp, data, &data_off);
-    if (result == -EAGAIN)
-      return;
-  }
+  // do it.
+  result = do_read_ops(&ctx, bp, data);
+  if (result == -EAGAIN)
+    return;
    
  done:
   // reply
   MOSDOpReply *reply = new MOSDOpReply(op, 0, osd->osdmap->get_epoch(), CEPH_OSD_FLAG_ACK); 
   reply->set_data(data);
-  reply->get_header().data_off = data_off;
+  reply->get_header().data_off = ctx.data_off;
   reply->set_result(result);
 
   if (result >= 0) {
@@ -915,7 +909,6 @@ void ReplicatedPG::op_read(MOSDOp *op)
       stat_object_temp_rd[soid].hit(now);  // hit temp.
   }
 
-  
   osd->messenger->send_message(reply, op->get_orig_source_inst());
   delete op;
 }
@@ -1685,7 +1678,7 @@ ReplicatedPG::ProjectedObjectInfo *ReplicatedPG::get_projected_object(pobject_t
     assert(r >= 0);
     pinfo->oi.decode(bv);
   } else {
-    pinfo->oi.poid = poid;
+    pinfo->oi.soid = poid;
     pinfo->exists = false;
     pinfo->size = 0;
   }

diff --git a/src/osd/ReplicatedPG.h b/src/osd/ReplicatedPG.h
index 4b958cd9b00c080473587058964a4b1868a62419..b173d93e36dd3f3962f82972016e21ad83cefec1 100644 (file)
--- a/src/osd/ReplicatedPG.h
+++ b/src/osd/ReplicatedPG.h
@@ -24,6 +24,21 @@ class MOSDSubOpReply;
 class ReplicatedPG : public PG {
 public:  
 
+  /*
+   * Capture all object state associated with an in-progress read.
+   */
+  struct ReadOpContext {
+    MOSDOp *op;
+    vector<ceph_osd_op>& ops;
+
+    object_info_t oi;
+    int data_off;        // FIXME: we may want to kill this msgr hint off at some point!
+
+    ReadOpContext(MOSDOp *_op, vector<ceph_osd_op>& _ops, sobject_t _soid) :
+      op(_op), ops(_ops), oi(_soid), data_off(0) {}
+  };
+
+
   /*
    * keep tabs on object modifications that are in flight.
    * we need to know the projected existence, size, snapset,
@@ -42,7 +57,6 @@ public:
     ProjectedObjectInfo() : ref(0), exists(false), size(0), oi(poid) {}
   };
 
-
   /*
    * gather state on the primary/head while replicating an osd op.
    */
@@ -122,13 +136,6 @@ public:
     }
   };
 
-  struct OpContext {
-    sobject_t& soid;
-    object_info_t& oi;
-    vector<ceph_osd_op>& ops;
-    OpContext(sobject_t& _soid, object_info_t& _oi,
-                vector<ceph_osd_op>& _ops) : soid(_soid), oi(_oi), ops(_ops) {}
-  };
 
 
 protected:
@@ -222,10 +229,7 @@ protected:
   void op_read(MOSDOp *op);
   void op_modify(MOSDOp *op);
 
-  int do_read_ops(MOSDOp *op, OpContext& ctx,
-                bufferlist::iterator& bp,
-                bufferlist& data,
-                int *data_off);
+  int do_read_ops(ReadOpContext *ctx, bufferlist::iterator& bp, bufferlist& data);
 
   void sub_op_modify(MOSDSubOp *op);
   void sub_op_modify_reply(MOSDSubOpReply *reply);

diff --git a/src/osd/osd_types.h b/src/osd/osd_types.h
index 161931c268819557eb05c1b1bbd8abeef7d1e829..810b20f555e156b06703c2f68de705c655a85912 100644 (file)
--- a/src/osd/osd_types.h
+++ b/src/osd/osd_types.h
@@ -726,7 +726,7 @@ inline ostream& operator<<(ostream& out, const SnapSet& cs) {
 #define OI_ATTR "_"
 
 struct object_info_t {
-  pobject_t poid;
+  sobject_t soid;
 
   eversion_t version, prior_version;
   osd_reqid_t last_reqid;
@@ -739,12 +739,12 @@ struct object_info_t {
   bufferlist truncate_info;  // bah.. messy layering.
 
   void encode(bufferlist& bl) const {
-    ::encode(poid, bl);
+    ::encode(soid, bl);
     ::encode(version, bl);
     ::encode(prior_version, bl);
     ::encode(last_reqid, bl);
     ::encode(mtime, bl);
-    if (poid.snap == CEPH_NOSNAP) {
+    if (soid.snap == CEPH_NOSNAP) {
       ::encode(snapset, bl);
       ::encode(wrlock_by, bl);
     } else
@@ -752,12 +752,12 @@ struct object_info_t {
     ::encode(truncate_info, bl);
   }
   void decode(bufferlist::iterator& bl) {
-    ::decode(poid, bl);
+    ::decode(soid, bl);
     ::decode(version, bl);
     ::decode(prior_version, bl);
     ::decode(last_reqid, bl);
     ::decode(mtime, bl);
-    if (poid.snap == CEPH_NOSNAP) {
+    if (soid.snap == CEPH_NOSNAP) {
       ::decode(snapset, bl);
       ::decode(wrlock_by, bl);
     } else
@@ -769,7 +769,7 @@ struct object_info_t {
     decode(p);
   }
 
-  object_info_t(pobject_t p) : poid(p) {}
+  object_info_t(sobject_t s) : soid(s) {}
   object_info_t(bufferlist& bl) {
     decode(bl);
   }
@@ -778,11 +778,11 @@ WRITE_CLASS_ENCODER(object_info_t)
 
 
 inline ostream& operator<<(ostream& out, const object_info_t& oi) {
-  out << oi.poid << "(" << oi.version
+  out << oi.soid << "(" << oi.version
       << " " << oi.last_reqid;
   if (oi.wrlock_by.tid)
     out << " wrlock_by=" << oi.wrlock_by;
-  if (oi.poid.snap == CEPH_NOSNAP)
+  if (oi.soid.snap == CEPH_NOSNAP)
     out << " " << oi.snapset << ")";
   else
     out << " " << oi.snaps << ")";