ReplicatedPG: follow the same finish path for failed copy ops

We don't necessarily want to respond to clients with a failure if
a copy got an error code. Instead, conditionally execute the success
path and always launch back into execute_ctx() when the copy has
stopped (either due to completion or failure).

Update the COPY_FROM section so it returns the CopyOp::rval (instead
of always zero) and only launches finish_copy() on success.

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

1 files changed, 29 insertions, 30 deletions

diff --git a/src/osd/ReplicatedPG.cc b/src/osd/ReplicatedPG.cc
index 4748c31203b..feee6de920f 100644
--- a/src/osd/ReplicatedPG.cc
+++ b/src/osd/ReplicatedPG.cc
@@ -3770,7 +3770,10 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops)
 	  result = -EINPROGRESS;
 	} else {
 	  // finish
-	  result = finish_copy(ctx);
+	  result = ctx->copy_op->rval;
+	  if (ctx->copy_op->rval >= 0) { //success!
+	    result = finish_copy(ctx);
+	  }
 	}
       }
       break;
@@ -4449,39 +4452,35 @@ void ReplicatedPG::process_copy_chunk(hobject_t oid, tid_t tid, int r)
   OpContext *ctx = cop->ctx;
   ObjectContextRef obc = ctx->obc;
   cop->objecter_tid = 0;
-  if (r < 0) {
-    copy_ops.erase(obc->obs.oi.soid);
-    obc->copyfrom_readside;
-    kick_object_context_blocked(obc);
-    reply_ctx(ctx, r);
-    return;
-  }
-  assert(cop->rval >= 0);
 
-  if (!cop->cursor.is_complete()) {
-    // write out what we have so far
-    vector<OSDOp> ops;
-    tid_t rep_tid = osd->get_tid();
-    osd_reqid_t reqid(osd->get_cluster_msgr_name(), 0, rep_tid);
-    OpContext *tctx = new OpContext(OpRequestRef(), reqid, ops, &obc->obs, obc->ssc, this);
-    tctx->mtime = ceph_clock_now(g_ceph_context);
-    RepGather *repop = new_repop(tctx, obc, rep_tid);
-
-    if (cop->temp_cursor.is_initial()) {
-      cop->temp_coll = get_temp_coll(&tctx->local_t);
-      cop->temp_oid = generate_temp_object();
-      repop->ctx->new_temp_oid = cop->temp_oid;
-    }
+  if (r >= 0) {
+    assert(cop->rval >= 0);
+
+    if (!cop->cursor.is_complete()) {
+      // write out what we have so far
+      vector<OSDOp> ops;
+      tid_t rep_tid = osd->get_tid();
+      osd_reqid_t reqid(osd->get_cluster_msgr_name(), 0, rep_tid);
+      OpContext *tctx = new OpContext(OpRequestRef(), reqid, ops, &obc->obs, obc->ssc, this);
+      tctx->mtime = ceph_clock_now(g_ceph_context);
+      RepGather *repop = new_repop(tctx, obc, rep_tid);
+
+      if (cop->temp_cursor.is_initial()) {
+	cop->temp_coll = get_temp_coll(&tctx->local_t);
+	cop->temp_oid = generate_temp_object();
+	repop->ctx->new_temp_oid = cop->temp_oid;
+      }
 
-    _write_copy_chunk(cop, &tctx->op_t);
+      _write_copy_chunk(cop, &tctx->op_t);
 
-    issue_repop(repop, repop->ctx->mtime);
-    eval_repop(repop);
-    repop->put();
+      issue_repop(repop, repop->ctx->mtime);
+      eval_repop(repop);
+      repop->put();
 
-    dout(10) << __func__ << " fetching more" << dendl;
-    _copy_some(obc, cop);
-    return;
+      dout(10) << __func__ << " fetching more" << dendl;
+      _copy_some(obc, cop);
+      return;
+    }
   }
 
   dout(20) << __func__ << " complete; committing" << dendl;