osd: generic async operation finisher handler

The COPY_FROM operation has a custom data path for restarting
itself after the read has completed. The CMPEXT operation will
need similar restart handling when running on EC pools so that
it can guard ops after an async read completes.

Signed-off-by: Jason Dillaman <dillaman@redhat.com>

diff --git a/src/osd/PrimaryLogPG.cc b/src/osd/PrimaryLogPG.cc
index 001e10fdc2c076b31a52af0ed74a914715a7518d..a90c426ecf790b566cfc8cab7f6bdcf45fb06fa2 100644 (file)
--- a/src/osd/PrimaryLogPG.cc
+++ b/src/osd/PrimaryLogPG.cc
@@ -273,30 +273,39 @@ void PrimaryLogPG::OpContext::finish_read(PrimaryLogPG *pg)
   }
 }
 
-class CopyFromCallback: public PrimaryLogPG::CopyCallback {
+struct OpFinisher {
+  OSDOp &osd_op;
+
+  OpFinisher(OSDOp &osd_op) : osd_op(osd_op) {
+  }
+  virtual ~OpFinisher() {
+    osd_op.op_finisher = nullptr;
+  }
+
+  virtual int execute() = 0;
+};
+
+class CopyFromCallback : public PrimaryLogPG::CopyCallback {
 public:
-  PrimaryLogPG::CopyResults *results;
-  int retval;
+  PrimaryLogPG::CopyResults *results = nullptr;
   PrimaryLogPG::OpContext *ctx;
-  explicit CopyFromCallback(PrimaryLogPG::OpContext *ctx_)
-    : results(NULL),
-      retval(0),
-      ctx(ctx_) {}
+  OSDOp &osd_op;
+
+  CopyFromCallback(PrimaryLogPG::OpContext *ctx, OSDOp &osd_op)
+    : ctx(ctx), osd_op(osd_op) {
+  }
   ~CopyFromCallback() override {}
 
   void finish(PrimaryLogPG::CopyCallbackResults results_) override {
     results = results_.get<1>();
     int r = results_.get<0>();
-    retval = r;
 
     // for finish_copyfrom
     ctx->user_at_version = results->user_version;
 
     if (r >= 0) {
       ctx->pg->execute_ctx(ctx);
-    }
-    ctx->copy_cb = NULL;
-    if (r < 0) {
+    } else {
       if (r != -ECANCELED) { // on cancel just toss it out; client resends
        if (ctx->op)
          ctx->pg->osd->reply_op_error(ctx->op, r);
@@ -314,8 +323,18 @@ public:
   uint64_t get_data_size() {
     return results->object_size;
   }
-  int get_result() {
-    return retval;
+};
+
+struct CopyFromFinisher : public OpFinisher {
+  CopyFromCallback *copy_from_callback;  
+
+  CopyFromFinisher(OSDOp &osd_op, CopyFromCallback *copy_from_callback)
+    : OpFinisher(osd_op), copy_from_callback(copy_from_callback) {
+  }
+
+  int execute() override {
+    copy_from_callback->ctx->pg->finish_copyfrom(copy_from_callback);
+    return 0;
   }
 };
 
@@ -3364,6 +3383,23 @@ void PrimaryLogPG::execute_ctx(OpContext *ctx)
   repop->put();
 }
 
+void PrimaryLogPG::close_op_ctx(OpContext *ctx) {
+  release_object_locks(ctx->lock_manager);
+
+  if (ctx->ops != nullptr) {
+    for (auto &osd_op : *ctx->ops) {
+      delete osd_op.op_finisher;
+    }
+  }
+  ctx->op_t.reset();
+
+  for (auto p = ctx->on_finish.begin(); p != ctx->on_finish.end();
+       ctx->on_finish.erase(p++)) {
+    (*p)();
+  }
+  delete ctx;
+}
+
 void PrimaryLogPG::reply_ctx(OpContext *ctx, int r)
 {
   if (ctx->op)
@@ -6549,7 +6585,7 @@ int PrimaryLogPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops)
                   src_oloc.hash,
                   src_snapid,
                   src_version);
-       if (!ctx->copy_cb) {
+       if (osd_op.op_finisher == nullptr) {
          // start
          pg_t raw_pg;
          get_osdmap()->object_locator_to_pg(src_name, src_oloc, raw_pg);
@@ -6561,8 +6597,8 @@ int PrimaryLogPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops)
            result = -EINVAL;
            break;
          }
-         CopyFromCallback *cb = new CopyFromCallback(ctx);
-         ctx->copy_cb = cb;
+         CopyFromCallback *cb = new CopyFromCallback(ctx, osd_op);
+         osd_op.op_finisher = new CopyFromFinisher(osd_op, cb);
          start_copy(cb, ctx->obc, src, src_oloc, src_version,
                     op.copy_from.flags,
                     false,
@@ -6571,9 +6607,8 @@ int PrimaryLogPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops)
          result = -EINPROGRESS;
        } else {
          // finish
-         assert(ctx->copy_cb->get_result() >= 0);
-         finish_copyfrom(ctx);
-         result = 0;
+         result = osd_op.op_finisher->execute();
+         assert(result == 0);
        }
       }
       break;
@@ -8159,12 +8194,12 @@ void PrimaryLogPG::_write_copy_chunk(CopyOpRef cop, PGTransaction *t)
   cop->temp_cursor = cop->cursor;
 }
 
-void PrimaryLogPG::finish_copyfrom(OpContext *ctx)
+void PrimaryLogPG::finish_copyfrom(CopyFromCallback *cb)
 {
+  OpContext *ctx = cb->ctx;
   dout(20) << "finish_copyfrom on " << ctx->obs->oi.soid << dendl;
-  ObjectState& obs = ctx->new_obs;
-  CopyFromCallback *cb = static_cast<CopyFromCallback*>(ctx->copy_cb);
 
+  ObjectState& obs = ctx->new_obs;
   if (obs.exists) {
     dout(20) << __func__ << ": exists, removing" << dendl;
     ctx->op_t->remove(obs.oi.soid);

diff --git a/src/osd/PrimaryLogPG.h b/src/osd/PrimaryLogPG.h
index 47ce60e048d754f6c404bab8e549d8060664521e..99b902d3e01eebf43c250a4bdac00c5be9e7c2b2 100644 (file)
--- a/src/osd/PrimaryLogPG.h
+++ b/src/osd/PrimaryLogPG.h
@@ -177,6 +177,7 @@ public:
   typedef boost::tuple<int, CopyResults*> CopyCallbackResults;
 
   friend class CopyFromCallback;
+  friend class CopyFromFinisher;
   friend class PromoteCallback;
 
   struct ProxyReadOp {
@@ -540,8 +541,6 @@ public:
 
     mempool::osd_pglog::vector<pair<osd_reqid_t, version_t> > extra_reqids;
 
-    CopyFromCallback *copy_cb;
-
     hobject_t new_temp_oid, discard_temp_oid;  ///< temp objects we should start/stop tracking
 
     list<std::function<void()>> on_applied;
@@ -600,7 +599,6 @@ public:
       data_off(0), reply(NULL), pg(_pg),
       num_read(0),
       num_write(0),
-      copy_cb(NULL),
       sent_reply(false),
       async_read_result(0),
       inflightreads(0),
@@ -620,7 +618,6 @@ public:
       data_off(0), reply(NULL), pg(_pg),
       num_read(0),
       num_write(0),
-      copy_cb(NULL),
       async_read_result(0),
       inflightreads(0),
       lock_type(ObjectContext::RWState::RWNONE) {}
@@ -794,16 +791,7 @@ protected:
    *
    * @param ctx [in] ctx to clean up
    */
-  void close_op_ctx(OpContext *ctx) {
-    release_object_locks(ctx->lock_manager);
-    ctx->op_t.reset();
-    for (auto p = ctx->on_finish.begin();
-        p != ctx->on_finish.end();
-        ctx->on_finish.erase(p++)) {
-      (*p)();
-    }
-    delete ctx;
-  }
+  void close_op_ctx(OpContext *ctx);
 
   /**
    * Releases locks
@@ -1286,7 +1274,7 @@ protected:
     return size;
   }
   void _copy_some(ObjectContextRef obc, CopyOpRef cop);
-  void finish_copyfrom(OpContext *ctx);
+  void finish_copyfrom(CopyFromCallback *cb);
   void finish_promote(int r, CopyResults *results, ObjectContextRef obc);
   void cancel_copy(CopyOpRef cop, bool requeue);
   void cancel_copy_ops(bool requeue);

diff --git a/src/osd/osd_types.h b/src/osd/osd_types.h
index 503720955d849b7f12810c6c0a5a2b015ffe0b41..8834be0d22f18f6f4ed28c4c8b0aea611f6b90ef 100644 (file)
--- a/src/osd/osd_types.h
+++ b/src/osd/osd_types.h
@@ -4869,6 +4869,7 @@ struct ScrubMap {
 WRITE_CLASS_ENCODER(ScrubMap::object)
 WRITE_CLASS_ENCODER(ScrubMap)
 
+struct OpFinisher;
 
 struct OSDOp {
   ceph_osd_op op;
@@ -4877,6 +4878,8 @@ struct OSDOp {
   bufferlist indata, outdata;
   errorcode32_t rval;
 
+  OpFinisher *op_finisher = nullptr;
+
   OSDOp() : rval(0) {
     memset(&op, 0, sizeof(ceph_osd_op));
   }