objecter: separate out linger_read() and linger_mutate()

A watch is a mutation, while a notify is a read.  The mutations need to
pass in a proper snap context to be fully correct.

Also, make the WRITE flag implicit so the caller doesn't need to pass it
in.

Signed-off-by: Sage Weil <sage@inktank.com>

3 files changed, 59 insertions, 17 deletions

diff --git a/src/librados/IoCtxImpl.cc b/src/librados/IoCtxImpl.cc
index 5f09dd47248..626999fa75d 100644
--- a/src/librados/IoCtxImpl.cc
+++ b/src/librados/IoCtxImpl.cc
@@ -1407,10 +1407,10 @@ int librados::IoCtxImpl::watch(const object_t& oid, uint64_t ver,
   prepare_assert_ops(&wr);
   wr.watch(*cookie, ver, 1);
   bufferlist bl;
-  wc->linger_id = objecter->linger(
-    oid, oloc, wr, snap_seq, bl, NULL,
-    CEPH_OSD_FLAG_WRITE,
-    NULL, onfinish, &objver);
+  wc->linger_id = objecter->linger_mutate(oid, oloc, wr,
+					  snapc, ceph_clock_now(NULL), bl,
+					  0,
+					  NULL, onfinish, &objver);
   lock->Unlock();
 
   mylock.Lock();
@@ -1500,8 +1500,8 @@ int librados::IoCtxImpl::notify(const object_t& oid, uint64_t ver, bufferlist& b
   ::encode(timeout, inbl);
   ::encode(bl, inbl);
   rd.notify(cookie, ver, inbl);
-  wc->linger_id = objecter->linger(oid, oloc, rd, snap_seq, inbl, NULL,
-				   0, onack, NULL, &objver);
+  wc->linger_id = objecter->linger_read(oid, oloc, rd, snap_seq, inbl, NULL, 0,
+					onack, &objver);
   lock->Unlock();
 
   mylock.Lock();
diff --git a/src/osdc/Objecter.cc b/src/osdc/Objecter.cc
index 21d9df7f3d6..2cb6e555e77 100644
--- a/src/osdc/Objecter.cc
+++ b/src/osdc/Objecter.cc
@@ -265,6 +265,8 @@ void Objecter::send_linger(LingerOp *info)
 		 onack, oncommit,
 		 info->pobjver);
   o->snapid = info->snap;
+  o->snapc = info->snapc;
+  o->mtime = info->mtime;
 
   // do not resend this; we will send a new op to reregister
   o->should_resend = false;
@@ -335,11 +337,43 @@ void Objecter::unregister_linger(uint64_t linger_id)
   }
 }
 
-tid_t Objecter::linger(const object_t& oid, const object_locator_t& oloc, 
-		       ObjectOperation& op,
-		       snapid_t snap, bufferlist& inbl, bufferlist *poutbl, int flags,
-		       Context *onack, Context *onfinish,
-		       eversion_t *objver)
+tid_t Objecter::linger_mutate(const object_t& oid, const object_locator_t& oloc,
+			      ObjectOperation& op,
+			      const SnapContext& snapc, utime_t mtime,
+			      bufferlist& inbl, int flags,
+			      Context *onack, Context *oncommit,
+			      eversion_t *objver)
+{
+  LingerOp *info = new LingerOp;
+  info->oid = oid;
+  info->oloc = oloc;
+  if (info->oloc.key == oid)
+    info->oloc.key.clear();
+  info->snapc = snapc;
+  info->mtime = mtime;
+  info->flags = flags | CEPH_OSD_FLAG_WRITE;
+  info->ops = op.ops;
+  info->inbl = inbl;
+  info->poutbl = NULL;
+  info->pobjver = objver;
+  info->on_reg_ack = onack;
+  info->on_reg_commit = oncommit;
+
+  info->linger_id = ++max_linger_id;
+  linger_ops[info->linger_id] = info;
+
+  logger->set(l_osdc_linger_active, linger_ops.size());
+
+  send_linger(info);
+
+  return info->linger_id;
+}
+
+tid_t Objecter::linger_read(const object_t& oid, const object_locator_t& oloc,
+			    ObjectOperation& op,
+			    snapid_t snap, bufferlist& inbl, bufferlist *poutbl, int flags,
+			    Context *onfinish,
+			    eversion_t *objver)
 {
   LingerOp *info = new LingerOp;
   info->oid = oid;
@@ -352,7 +386,6 @@ tid_t Objecter::linger(const object_t& oid, const object_locator_t& oloc,
   info->inbl = inbl;
   info->poutbl = poutbl;
   info->pobjver = objver;
-  info->on_reg_ack = onack;
   info->on_reg_commit = onfinish;
 
   info->linger_id = ++max_linger_id;
diff --git a/src/osdc/Objecter.h b/src/osdc/Objecter.h
index 9ff02f6ab93..a31cac2a03c 100644
--- a/src/osdc/Objecter.h
+++ b/src/osdc/Objecter.h
@@ -803,6 +803,9 @@ public:
     vector<int> acting;
 
     snapid_t snap;
+    SnapContext snapc;
+    utime_t mtime;
+
     int flags;
     vector<OSDOp> ops;
     bufferlist inbl;
@@ -1069,11 +1072,17 @@ private:
     o->out_rval.swap(op.out_rval);
     return op_submit(o);
   }
-  tid_t linger(const object_t& oid, const object_locator_t& oloc, 
-	       ObjectOperation& op,
-	       snapid_t snap, bufferlist& inbl, bufferlist *poutbl, int flags,
-               Context *onack, Context *onfinish,
-               eversion_t *objver);
+  tid_t linger_mutate(const object_t& oid, const object_locator_t& oloc,
+		      ObjectOperation& op,
+		      const SnapContext& snapc, utime_t mtime,
+		      bufferlist& inbl, int flags,
+		      Context *onack, Context *onfinish,
+		      eversion_t *objver);
+  tid_t linger_read(const object_t& oid, const object_locator_t& oloc,
+		    ObjectOperation& op,
+		    snapid_t snap, bufferlist& inbl, bufferlist *poutbl, int flags,
+		    Context *onack,
+		    eversion_t *objver);
   void unregister_linger(uint64_t linger_id);
 
   /**