osdc/Objecter: unwatch is a mutation, not a read

This was causing librados to unblock after the ACK on unwatch, which meant
that librbd users raced and tried to delete the image before the unwatch
change was committed..and got EBUSY.  See #3958.

The watch operation has a similar problem.

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

diff --git a/src/librados/IoCtxImpl.cc b/src/librados/IoCtxImpl.cc
index 800e27f90b6e7aae18e522011508d534019655d4..5f09dd4724810eab7dfce0aeeae11c01197bfd09 100644 (file)
--- a/src/librados/IoCtxImpl.cc
+++ b/src/librados/IoCtxImpl.cc
@@ -1392,7 +1392,7 @@ void librados::IoCtxImpl::set_sync_op_version(eversion_t& ver)
 int librados::IoCtxImpl::watch(const object_t& oid, uint64_t ver,
                               uint64_t *cookie, librados::WatchCtx *ctx)
 {
-  ::ObjectOperation rd;
+  ::ObjectOperation wr;
   Mutex mylock("IoCtxImpl::watch::mylock");
   Cond cond;
   bool done;
@@ -1404,11 +1404,11 @@ int librados::IoCtxImpl::watch(const object_t& oid, uint64_t ver,
 
   WatchContext *wc = new WatchContext(this, oid, ctx);
   client->register_watcher(wc, cookie);
-  prepare_assert_ops(&rd);
-  rd.watch(*cookie, ver, 1);
+  prepare_assert_ops(&wr);
+  wr.watch(*cookie, ver, 1);
   bufferlist bl;
   wc->linger_id = objecter->linger(
-    oid, oloc, rd, snap_seq, bl, NULL,
+    oid, oloc, wr, snap_seq, bl, NULL,
     CEPH_OSD_FLAG_WRITE,
     NULL, onfinish, &objver);
   lock->Unlock();
@@ -1452,16 +1452,16 @@ int librados::IoCtxImpl::unwatch(const object_t& oid, uint64_t cookie)
   Cond cond;
   bool done;
   int r;
-  Context *onack = new C_SafeCond(&mylock, &cond, &done, &r);
+  Context *oncommit = new C_SafeCond(&mylock, &cond, &done, &r);
   eversion_t ver;
   lock->Lock();
 
   client->unregister_watcher(cookie);
 
-  ::ObjectOperation rd;
-  prepare_assert_ops(&rd);
-  rd.watch(cookie, 0, 0);
-  objecter->read(oid, oloc, rd, snap_seq, &outbl, 0, onack, &ver);
+  ::ObjectOperation wr;
+  prepare_assert_ops(&wr);
+  wr.watch(cookie, 0, 0);
+  objecter->mutate(oid, oloc, wr, snapc, ceph_clock_now(client->cct), 0, NULL, oncommit, &ver);
   lock->Unlock();
 
   mylock.Lock();