ReplicatedPG: copy: use CopyCallback instead of CopyOp in OpContext

In order to make this happen, we make the switch to generate the complete
transaction in the generic copy code and save it into the Callback. Then
in finish_copy() we just take that transaction and prepend it to the existing
transaction.
With that change, and by making use of the existing CopyCallback data,
we no longer need to access the CopyOp from the OpContext, so we can remove it.
Hurray, the pipelines are now independent!

Signed-off-by: Greg Farnum <greg@inktank.com>

diff --git a/src/osd/ReplicatedPG.cc b/src/osd/ReplicatedPG.cc
index 8d8dde1b36534ea978287740ed7c3d86cb398c64..35e28c18361b866940899d5a3573481aeaeab7a3 100644 (file)
--- a/src/osd/ReplicatedPG.cc
+++ b/src/osd/ReplicatedPG.cc
@@ -3752,7 +3752,7 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops)
          result = -EINVAL;
          goto fail;
        }
-       if (!ctx->copy_op) {
+       if (!ctx->copy_cb) {
          // start
          pg_t raw_pg;
          get_osdmap()->object_locator_to_pg(src_name, src_oloc, raw_pg);
@@ -3766,15 +3766,16 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops)
          }
          hobject_t temp_target = generate_temp_object();
          CopyFromCallback *cb = new CopyFromCallback(ctx, temp_target);
-         result = start_copy(ctx, cb, ctx->obc, src, src_oloc, src_version,
+         ctx->copy_cb = cb;
+         result = start_copy(cb, ctx->obc, src, src_oloc, src_version,
                              temp_target);
          if (result < 0)
            goto fail;
          result = -EINPROGRESS;
        } else {
          // finish
-         result = ctx->copy_op->rval;
-         if (ctx->copy_op->rval >= 0) { //success!
+         result = ctx->copy_cb->get_result();
+         if (result >= 0) { //success!
            result = finish_copy(ctx);
          }
        }
@@ -4382,12 +4383,12 @@ struct C_Copyfrom : public Context {
   }
 };
 
-int ReplicatedPG::start_copy(OpContext *ctx, CopyCallback *cb, ObjectContextRef obc,
+int ReplicatedPG::start_copy(CopyCallback *cb, ObjectContextRef obc,
                             hobject_t src, object_locator_t oloc, version_t version,
                             const hobject_t& temp_dest_oid)
 {
-  const hobject_t& dest = ctx->obs->oi.soid;
-  dout(10) << __func__ << " " << dest << " ctx " << ctx
+  const hobject_t& dest = obc->obs.oi.soid;
+  dout(10) << __func__ << " " << dest
           << " from " << src << " " << oloc << " v" << version
           << dendl;
 
@@ -4401,7 +4402,6 @@ int ReplicatedPG::start_copy(OpContext *ctx, CopyCallback *cb, ObjectContextRef
 
   CopyOpRef cop(new CopyOp(cb, obc, src, oloc, version, temp_dest_oid));
   copy_ops[dest] = cop;
-  ctx->copy_op = cop;
   ++obc->copyfrom_readside;
 
   _copy_some(obc, cop);
@@ -4482,6 +4482,11 @@ void ReplicatedPG::process_copy_chunk(hobject_t oid, tid_t tid, int r)
       dout(10) << __func__ << " fetching more" << dendl;
       _copy_some(obc, cop);
       return;
+    } else {
+      ObjectStore::Transaction t;
+      _build_finish_copy_transaction(cop, t);
+      cop->cb->copy_complete_ops(t);
+      cop->cb->set_data_size(cop->temp_cursor.data_offset);
     }
   }
 
@@ -4541,28 +4546,27 @@ void ReplicatedPG::_build_finish_copy_transaction(CopyOpRef cop,
 
 int ReplicatedPG::finish_copy(OpContext *ctx)
 {
-  CopyOpRef cop = ctx->copy_op;
   ObjectState& obs = ctx->new_obs;
-  ObjectStore::Transaction& t = ctx->op_t;
+  CopyFromCallback *cb = static_cast<CopyFromCallback*>(ctx->copy_cb);
 
   if (!ctx->obs->exists) {
     ctx->delta_stats.num_objects++;
     obs.exists = true;
   }
-  if (cop->temp_cursor.is_initial()) {
-    ctx->discard_temp_oid = cop->temp_oid;
+  if (cb->is_temp_obj_used()) {
+    ctx->discard_temp_oid = cb->temp_obj;
   }
-
-  _build_finish_copy_transaction(cop, t);
+  ctx->op_t.swap(cb->final_tx);
+  ctx->op_t.append(cb->final_tx);
 
   interval_set<uint64_t> ch;
   if (obs.oi.size > 0)
     ch.insert(0, obs.oi.size);
   ctx->modified_ranges.union_of(ch);
 
-  if (cop->cursor.data_offset != obs.oi.size) {
+  if (cb->get_data_size() != obs.oi.size) {
     ctx->delta_stats.num_bytes -= obs.oi.size;
-    obs.oi.size = cop->cursor.data_offset;
+    obs.oi.size = cb->get_data_size();
     ctx->delta_stats.num_bytes += obs.oi.size;
   }
   ctx->delta_stats.num_wr++;

diff --git a/src/osd/ReplicatedPG.h b/src/osd/ReplicatedPG.h
index 00611104555ab08ce44aa01c51302a6c3f127990..d5ef9b26ca3f1434f9a809cf5e01f60ad84ea8d3 100644 (file)
--- a/src/osd/ReplicatedPG.h
+++ b/src/osd/ReplicatedPG.h
@@ -182,7 +182,6 @@ public:
       if (r >= 0) {
        ctx->pg->execute_ctx(ctx);
       }
-      ctx->copy_op.reset();
       ctx->copy_cb = NULL;
       if (r < 0) {
        if (r == -ECANCELED) { // toss it out; client resends
@@ -195,9 +194,10 @@ public:
   public:
     OpContext *ctx;
     hobject_t temp_obj;
+    ObjectStore::Transaction final_tx;
     CopyFromCallback(OpContext *ctx_, const hobject_t& temp_obj_) :
       ctx(ctx_), temp_obj(temp_obj_) {}
-    void copy_complete_ops(ObjectStore::Transaction& t) {}
+    void copy_complete_ops(ObjectStore::Transaction& t) { final_tx.swap(t); }
     ~CopyFromCallback() {}
   };
   friend class CopyFromCallback;
@@ -375,7 +375,7 @@ public:
     int num_read;    ///< count read ops
     int num_write;   ///< count update ops
 
-    CopyOpRef copy_op;
+    CopyCallback *copy_cb;
 
     hobject_t new_temp_oid, discard_temp_oid;  ///< temp objects we should start/stop tracking
 
@@ -392,7 +392,8 @@ public:
       current_osd_subop_num(0),
       data_off(0), reply(NULL), pg(_pg),
       num_read(0),
-      num_write(0) {
+      num_write(0),
+      copy_cb(NULL) {
       if (_ssc) {
        new_snapset = _ssc->snapset;
        snapset = &_ssc->snapset;
@@ -801,7 +802,7 @@ protected:
   // -- copyfrom --
   map<hobject_t, CopyOpRef> copy_ops;
 
-  int start_copy(OpContext *ctx, CopyCallback *cb, ObjectContextRef obc, hobject_t src,
+  int start_copy(CopyCallback *cb, ObjectContextRef obc, hobject_t src,
                  object_locator_t oloc, version_t version,
                  const hobject_t& temp_dest_oid);
   void process_copy_chunk(hobject_t oid, tid_t tid, int r);