From 2b5e61ca92516c0ab15cf56c26f0bdeb15059b97 Mon Sep 17 00:00:00 2001
From: Yehuda Sadeh <yehuda@hq.newdream.net>
Date: Fri, 24 Sep 2010 14:19:14 -0700
Subject: [PATCH] osd: send notification id

---
 src/osd/ReplicatedPG.cc | 20 ++++++++++++--------
 src/osd/Watch.h         | 12 +++++++++++-
 2 files changed, 23 insertions(+), 9 deletions(-)

diff --git a/src/osd/ReplicatedPG.cc b/src/osd/ReplicatedPG.cc
index 410116a703ef8..02642c3c343ad 100644
--- a/src/osd/ReplicatedPG.cc
+++ b/src/osd/ReplicatedPG.cc
@@ -1107,7 +1107,7 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops,
 	    dout(0) << " found session, sending notification" << dendl;
 	    notif->add_watcher(oi_iter->first, Watch::WATCHER_NOTIFIED); // adding before send_message to avoid race
 
-	    MWatchNotify *notify_msg = new MWatchNotify(w.cookie, w.ver);
+	    MWatchNotify *notify_msg = new MWatchNotify(w.cookie, w.ver, notif->id);
 	    osd->client_messenger->send_message(notify_msg, session->con);
 	  } else {
 	    notif->add_watcher(oi_iter->first, Watch::WATCHER_PENDING);
@@ -1137,13 +1137,15 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops,
 	watch_info_t& wi = oi_iter->second;
 	wi.ver = op.watch.ver;
 
-	map<entity_name_t, bool>::iterator pending_iter =
-			obc->pending_watchers.find(ctx->op->get_source());
-	if (pending_iter != obc->pending_watchers.end()) {
-	  obc->pending_watchers.erase(pending_iter);
-	  if (obc->pending_watchers.empty()) {
-	    dout(0) << "got the last reply from pending watchers, can send response now" << dendl;
-	  }
+	Watch::Notification *notif = osd->watch->get_notif(op.watch.cookie);
+	if (!notif) {
+	  result = -EINVAL;
+	  break;
+	}
+
+	entity_name_t source = ctx->op->get_source();
+	if (osd->watch->ack_notification(source, notif)) {
+	  dout(0) << "got the last reply from pending watchers, can send response now" << dendl;
 	}
       }
       break;
@@ -1347,6 +1349,8 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops,
 	  map<entity_name_t, watch_info_t>::iterator oi_iter = oi.watchers.find(entity);
 	  if (oi_iter !=  oi.watchers.end())
             oi.watchers.erase(entity);
+
+	  /* we don't session->put() here because we already did it above */
         }
 
 
diff --git a/src/osd/Watch.h b/src/osd/Watch.h
index 3d5e2a93fd549..9a8e8e712cb1b 100644
--- a/src/osd/Watch.h
+++ b/src/osd/Watch.h
@@ -23,6 +23,7 @@
 
 /* keeps track and accounts sessions, watchers and notifiers */
 class Watch {
+  uint64_t notif_id;
 
 public:
   enum WatcherState {
@@ -33,6 +34,7 @@ public:
   struct Notification {
     std::map<entity_name_t, WatcherState> watchers;
     entity_name_t name;
+    uint64_t id;
 
     void add_watcher(const entity_name_t& name, WatcherState state) { watchers[name] = state; }
 
@@ -42,18 +44,26 @@ public:
 private:
   std::multimap<entity_name_t, Notification *> notifs;
   std::multimap<entity_name_t, Notification *> wtn; /* watchers to notifications */
+  std::map<uint64_t, Notification *> itn; /* notif_id to notifications */
 
   std::map<OSD::Session *, entity_name_t> ste; /* sessions to entities */
   std::map<entity_name_t, OSD::Session *> ets; /* entities to sessions */
 
 public:
 
-  Watch() {}
+  Watch() : notif_id(0) {}
 
   void register_session(OSD::Session *session, entity_name_t& name);
   void remove_session(OSD::Session *session);
   void add_notification(Notification *notif);
   bool ack_notification(entity_name_t& watcher, Notification *notif);
+
+  Notification *get_notif(uint64_t id) {
+    map<uint64_t, Notification *>::iterator iter = itn.find(id);
+    if (iter != itn.end())
+      return iter->second;
+    return NULL;
+  }
 };
 
 
-- 
2.39.5