osd: allow deletion of objects with watchers

If we delete an object with a watcher, forcefully disconnect the watcher
instead of preventing the deletion with EBUSY.

Fixes: #2339
Signed-off-by: Sage Weil <sage@redhat.com>

diff --git a/src/osd/ReplicatedPG.cc b/src/osd/ReplicatedPG.cc
index 8564a612602c6e922766fd2bb0592f0c578223ab..a398777c5e4a8766213cac8d606ca3deeb027b4e 100644 (file)
--- a/src/osd/ReplicatedPG.cc
+++ b/src/osd/ReplicatedPG.cc
@@ -4243,12 +4243,7 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops)
     case CEPH_OSD_OP_DELETE:
       ++ctx->num_write;
       tracepoint(osd, do_osd_op_pre_delete, soid.oid.name.c_str(), soid.snap.val);
-      if (ctx->obc->obs.oi.watchers.size()) {
-       // Cannot delete an object with watchers
-       result = -EBUSY;
-      } else {
-       result = _delete_oid(ctx, ctx->ignore_cache);
-      }
+      result = _delete_oid(ctx, ctx->ignore_cache);
       break;
 
     case CEPH_OSD_OP_CLONERANGE:
@@ -5036,6 +5031,17 @@ inline int ReplicatedPG::_delete_oid(OpContext *ctx, bool no_whiteout)
   }
   oi.size = 0;
 
+  // disconnect all watchers
+  for (map<pair<uint64_t, entity_name_t>, watch_info_t>::iterator p =
+        oi.watchers.begin();
+       p != oi.watchers.end();
+       ++p) {
+    dout(20) << __func__ << " will disconnect watcher " << p->first << dendl;
+    ctx->watch_disconnects.push_back(
+      OpContext::watch_disconnect_t(p->first.first, p->first.second, true));
+  }
+  oi.watchers.clear();
+
   // cache: cache: set whiteout on delete?
   if (pool.info.cache_mode != pg_pool_t::CACHEMODE_NONE && !no_whiteout) {
     dout(20) << __func__ << " setting whiteout on " << soid << dendl;

diff --git a/src/test/librados/watch_notify.cc b/src/test/librados/watch_notify.cc
index c81311e4387a793587726069d04c26358726e489..3598c139f52fcef031b06fd5c1be19b3972c02f0 100644 (file)
--- a/src/test/librados/watch_notify.cc
+++ b/src/test/librados/watch_notify.cc
@@ -216,10 +216,36 @@ TEST_F(LibRadosWatchNotifyECPP, WatchNotifyTimeout) {
 
 // --
 
+TEST_F(LibRadosWatchNotify, Watch2Delete) {
+  notify_io = ioctx;
+  notify_oid = "foo";
+  notify_err = 0;
+  char buf[128];
+  memset(buf, 0xcc, sizeof(buf));
+  ASSERT_EQ(0, rados_write(ioctx, notify_oid, buf, sizeof(buf), 0));
+  uint64_t handle;
+  ASSERT_EQ(0,
+           rados_watch2(ioctx, notify_oid, &handle,
+                        watch_notify2_test_cb,
+                        watch_notify2_test_errcb, NULL));
+  ASSERT_EQ(0, rados_remove(ioctx, notify_oid));
+  int left = 180;
+  std::cout << "waiting up to " << left << " for disconnect notification ..."
+           << std::endl;
+  while (notify_err == 0 && --left) {
+    sleep(1);
+  }
+  ASSERT_TRUE(left > 0);
+  ASSERT_EQ(-ENOTCONN, notify_err);
+  ASSERT_EQ(-ENOTCONN, rados_watch_check(ioctx, handle));
+  rados_unwatch2(ioctx, handle);
+}
+
 TEST_F(LibRadosWatchNotify, Watch2Timeout) {
   notify_io = ioctx;
   notify_oid = "foo";
   notify_cookies.clear();
+  notify_err = 0;
   char buf[128];
   memset(buf, 0xcc, sizeof(buf));
   ASSERT_EQ(0, rados_write(ioctx, notify_oid, buf, sizeof(buf), 0));