crimson/osd: dispatch MWatchNotify on watch removal.

Signed-off-by: Radoslaw Zarzynski <rzarzyns@redhat.com>

diff --git a/src/crimson/osd/ops_executer.cc b/src/crimson/osd/ops_executer.cc
index b1638d901985b06afbd7293c35c240bf7b25b8af..d332bbe807d8f570ab7006ef1b4c12ff377f1f8e 100644 (file)
--- a/src/crimson/osd/ops_executer.cc
+++ b/src/crimson/osd/ops_executer.cc
@@ -187,6 +187,8 @@ OpsExecuter::watch_errorator::future<> OpsExecuter::do_op_watch_subop_unwatch(
   ObjectState& os,
   ceph::os::Transaction& txn)
 {
+  logger().info("{}", __func__);
+
   struct disconnect_ctx_t {
     ObjectContext::watch_key_t key;
     bool send_disconnect{ false };

diff --git a/src/crimson/osd/watch.cc b/src/crimson/osd/watch.cc
index b5d13fd1ecaad90cb7bb32804d619e75f9c527ce..68eea9277d71ad6037f12d5eb77d238e7789c000 100644 (file)
--- a/src/crimson/osd/watch.cc
+++ b/src/crimson/osd/watch.cc
@@ -66,6 +66,42 @@ seastar::future<> Watch::notify_ack(
   });
 }
 
+seastar::future<> Watch::send_disconnect_msg()
+{
+  if (!is_connected()) {
+    return seastar::now();
+  }
+  ceph::bufferlist empty;
+  return conn->send(make_message<MWatchNotify>(
+    winfo.cookie,
+    0,
+    0,
+    CEPH_WATCH_EVENT_DISCONNECT,
+    empty));
+}
+
+void Watch::discard_state()
+{
+  ceph_assert(obc);
+  in_progress_notifies.clear();
+}
+
+seastar::future<> Watch::remove(const bool send_disconnect)
+{
+  logger().info("{}", __func__);
+  auto disconnected = send_disconnect ? send_disconnect_msg()
+                                      : seastar::now();
+  return std::move(disconnected).then([this] {
+    return seastar::do_for_each(in_progress_notifies,
+      [this_shared=shared_from_this()] (auto notify) {
+        return notify->remove_watcher(this_shared);
+      }).then([this] {
+        discard_state();
+        return seastar::now();
+      });
+    });
+}
+
 bool notify_reply_t::operator<(const notify_reply_t& rhs) const
 {
   // comparing std::pairs to emphasize our legacy. ceph-osd stores
@@ -78,6 +114,17 @@ bool notify_reply_t::operator<(const notify_reply_t& rhs) const
   return lhsp < rhsp;
 }
 
+seastar::future<> Notify::remove_watcher(WatchRef watch)
+{
+  if (discarded || complete) {
+    return seastar::now();
+  }
+  [[maybe_unused]] const auto num_removed = watchers.erase(watch);
+  assert(num_removed > 0);
+  return maybe_send_completion();
+}
+
+
 seastar::future<> Notify::complete_watcher(
   WatchRef watch,
   const ceph::bufferlist& reply_bl)
@@ -89,8 +136,7 @@ seastar::future<> Notify::complete_watcher(
     watch->get_watcher_gid(),
     watch->get_cookie(),
     reply_bl});
-  watchers.erase(watch);
-  return maybe_send_completion();
+  return remove_watcher(std::move(watch));
 }
 
 seastar::future<> Notify::maybe_send_completion()

diff --git a/src/crimson/osd/watch.h b/src/crimson/osd/watch.h
index bf9229028d404bdab1738c593de918f9946d0cff..6049e16cf4b5a2ab73fc4abd0d345f0f200e7d21 100644 (file)
--- a/src/crimson/osd/watch.h
+++ b/src/crimson/osd/watch.h
@@ -44,6 +44,8 @@ class Watch : public seastar::enable_shared_from_this<Watch> {
 
   seastar::future<> start_notify(NotifyRef);
   seastar::future<> send_notify_msg(NotifyRef);
+  seastar::future<> send_disconnect_msg();
+  void discard_state();
 
   friend Notify;
 
@@ -68,9 +70,7 @@ public:
     // NOP
   }
 
-  seastar::future<> remove(bool) {
-    return seastar::now();
-  }
+  seastar::future<> remove(bool send_disconnect);
 
   /// Call when notify_ack received on notify_id
   seastar::future<> notify_ack(
@@ -148,6 +148,7 @@ public:
     WatchIteratorT end,
     Args&&... args);
 
+  seastar::future<> remove_watcher(WatchRef watch);
   seastar::future<> complete_watcher(WatchRef watch,
                                      const ceph::bufferlist& reply_bl);
 };