objecter, librados: create Objecter::Op in two phases

(currently only in some librados operations)
First create the op, only then lock and submit so that we reduce lock
contention.

Signed-off-by: Yehuda Sadeh <yehuda@inktank.com>

diff --git a/src/librados/IoCtxImpl.cc b/src/librados/IoCtxImpl.cc
index 627bd3b2140339731080bf715e2458314e397fd7..fa5930ffe7e388738c57f123a86c47ef28c78f40 100644 (file)
--- a/src/librados/IoCtxImpl.cc
+++ b/src/librados/IoCtxImpl.cc
@@ -528,10 +528,11 @@ int librados::IoCtxImpl::operate(const object_t& oid, ::ObjectOperation *o,
 
   int op = o->ops[0].op.op;
   ldout(client->cct, 10) << ceph_osd_op_name(op) << " oid=" << oid << " nspace=" << oloc.nspace << dendl;
+  Objecter::Op *objecter_op = objecter->prepare_mutate_op(oid, oloc,
+                                                         *o, snapc, ut, 0,
+                                                         NULL, oncommit, &ver);
   lock->Lock();
-  objecter->mutate(oid, oloc,
-                  *o, snapc, ut, 0,
-                  NULL, oncommit, &ver);
+  objecter->op_submit(objecter_op);
   lock->Unlock();
 
   mylock.Lock();
@@ -562,10 +563,11 @@ int librados::IoCtxImpl::operate_read(const object_t& oid,
 
   int op = o->ops[0].op.op;
   ldout(client->cct, 10) << ceph_osd_op_name(op) << " oid=" << oid << " nspace=" << oloc.nspace << dendl;
+  Objecter::Op *objecter_op = objecter->prepare_read_op(oid, oloc,
+                                             *o, snap_seq, pbl, 0,
+                                             onack, &ver);
   lock->Lock();
-  objecter->read(oid, oloc,
-                  *o, snap_seq, pbl, 0,
-                  onack, &ver);
+  objecter->op_submit(objecter_op);
   lock->Unlock();
 
   mylock.Lock();
@@ -592,10 +594,11 @@ int librados::IoCtxImpl::aio_operate_read(const object_t &oid,
   c->io = this;
   c->pbl = pbl;
 
-  Mutex::Locker l(*lock);
-  objecter->read(oid, oloc,
+  Objecter::Op *objecter_op = objecter->prepare_read_op(oid, oloc,
                 *o, snap_seq, pbl, flags,
                 onack, &c->objver);
+  Mutex::Locker l(*lock);
+  objecter->op_submit(objecter_op);
   return 0;
 }
 

diff --git a/src/osdc/Objecter.h b/src/osdc/Objecter.h
index 4a1a54da4863023a71728f98c11ab3ee9394d0fa..fbc15c69dec897117b1d1bc7ad62c863afb39fbb 100644 (file)
--- a/src/osdc/Objecter.h
+++ b/src/osdc/Objecter.h
@@ -1481,12 +1481,12 @@ public:
 
 private:
   // low-level
-  tid_t op_submit(Op *op);
   tid_t _op_submit(Op *op);
   inline void unregister_op(Op *op);
 
   // public interface
- public:
+public:
+  tid_t op_submit(Op *op);
   bool is_active() {
     return !(ops.empty() && linger_ops.empty() && poolstat_ops.empty() && statfs_ops.empty());
   }
@@ -1548,7 +1548,7 @@ private:
   }
 
   // mid-level helpers
-  tid_t mutate(const object_t& oid, const object_locator_t& oloc, 
+  Op *prepare_mutate_op(const object_t& oid, const object_locator_t& oloc, 
               ObjectOperation& op,
               const SnapContext& snapc, utime_t mtime, int flags,
               Context *onack, Context *oncommit, version_t *objver = NULL) {
@@ -1556,9 +1556,16 @@ private:
     o->priority = op.priority;
     o->mtime = mtime;
     o->snapc = snapc;
+    return o;
+  }
+  tid_t mutate(const object_t& oid, const object_locator_t& oloc, 
+              ObjectOperation& op,
+              const SnapContext& snapc, utime_t mtime, int flags,
+              Context *onack, Context *oncommit, version_t *objver = NULL) {
+    Op *o = prepare_mutate_op(oid, oloc, op, snapc, mtime, flags, onack, oncommit, objver);
     return op_submit(o);
   }
-  tid_t read(const object_t& oid, const object_locator_t& oloc,
+  Op *prepare_read_op(const object_t& oid, const object_locator_t& oloc,
             ObjectOperation& op,
             snapid_t snapid, bufferlist *pbl, int flags,
             Context *onack, version_t *objver = NULL) {
@@ -1569,6 +1576,13 @@ private:
     o->out_bl.swap(op.out_bl);
     o->out_handler.swap(op.out_handler);
     o->out_rval.swap(op.out_rval);
+    return o;
+  }
+  tid_t read(const object_t& oid, const object_locator_t& oloc,
+            ObjectOperation& op,
+            snapid_t snapid, bufferlist *pbl, int flags,
+            Context *onack, version_t *objver = NULL) {
+    Op *o = prepare_read_op(oid, oloc, op, snapid, pbl, flags, onack, objver);
     return op_submit(o);
   }
   tid_t pg_read(uint32_t hash, object_locator_t oloc,