From 7272bfec1aaf42f3097a5bae595c323efdc22604 Mon Sep 17 00:00:00 2001
From: Yehuda Sadeh <yehuda@hq.newdream.net>
Date: Wed, 27 Oct 2010 16:51:35 -0700
Subject: [PATCH] osd: send notify response from reset handler if needed

---
 src/osd/OSD.cc          | 28 ++++++++++++++++++++++++++++
 src/osd/OSD.h           | 10 ++++++++++
 src/osd/ReplicatedPG.cc | 17 ++++++-----------
 3 files changed, 44 insertions(+), 11 deletions(-)

diff --git a/src/osd/OSD.cc b/src/osd/OSD.cc
index 63693b11daa76..f0d0a8d36bfec 100644
--- a/src/osd/OSD.cc
+++ b/src/osd/OSD.cc
@@ -1620,6 +1620,21 @@ void OSD::put_object_context(void *_obc, ceph_object_layout& layout)
   pg->unlock();
 }
 
+void OSD::ack_notification(entity_name_t& name, void *_notif)
+{
+  Watch::Notification *notif = (Watch::Notification *)_notif;
+  if (watch->ack_notification(name, notif)) {
+	dout(0) << "got the last reply from pending watchers, can send response now" << dendl;
+	MWatchNotify *reply = notif->reply; // new MWatchNotify(notif->cookie, wi.ver, notif->id, WATCH_NOTIFY_COMPLETE);
+	client_messenger->send_message(reply, notif->session->con);
+	notif->session->put();
+	watch->remove_notification(notif);
+	if (notif->timeout)
+	  watch_timer.cancel_event(notif->timeout);
+	delete notif;
+  }
+}
+
 bool OSD::ms_handle_reset(Connection *con)
 {
   dout(0) << "OSD::ms_handle_reset()" << dendl;
@@ -1668,6 +1683,19 @@ bool OSD::ms_handle_reset(Connection *con)
     put_object_context(obc, oiter->second);
   }
 
+  watch_lock.Lock();
+
+  map<void *, entity_name_t>::iterator notif_iter;
+  for (notif_iter = session->notifs.begin(); notif_iter != session->notifs.end(); ++notif_iter) {
+    Watch::Notification *notif = (Watch::Notification *)notif_iter->first;
+    entity_name_t& dest = notif_iter->second;
+    dout(0) << "ms_handle_reset: ack notification for notif=" << (void *)notif << " entity=" << dest << dendl;
+    ack_notification(dest, notif);
+  }
+  session->notifs.clear();
+
+  watch_lock.Unlock();
+
   return true;
 }
 
diff --git a/src/osd/OSD.h b/src/osd/OSD.h
index efb8d0f62ceda..6eacc6adb76aa 100644
--- a/src/osd/OSD.h
+++ b/src/osd/OSD.h
@@ -216,9 +216,18 @@ public:
     epoch_t last_sent_epoch;
     Connection *con;
     std::map<void *, ceph_object_layout> watches;
+    std::map<void *, entity_name_t> notifs;
 
     Session() : last_sent_epoch(0), con(0) {}
     ~Session() { if (con) con->put(); }
+    void add_notif(void *n, entity_name_t& name) {
+      notifs[n] = name;
+    }
+    void del_notif(void *n) {
+      std::map<void *, entity_name_t>::iterator iter = notifs.find(n);
+      if (iter != notifs.end())
+        notifs.erase(iter);
+    }
   };
 
 private:
@@ -987,6 +996,7 @@ public:
   PG *lookup_lock_pg(pg_t pgid);
   ReplicatedPG *get_pg(void *_obc, ceph_object_layout& layout);
   void put_object_context(void *_obc, ceph_object_layout& layout);
+  void ack_notification(entity_name_t& peer_addr, void *notif);
   Mutex watch_lock;
   SafeTimer watch_timer;
   void handle_notify_timeout(void *notif);
diff --git a/src/osd/ReplicatedPG.cc b/src/osd/ReplicatedPG.cc
index 7434077d83e5e..cdcc22c2eb5d0 100644
--- a/src/osd/ReplicatedPG.cc
+++ b/src/osd/ReplicatedPG.cc
@@ -1129,6 +1129,8 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops,
 	    session = iter->second;
 	    dout(0) << " found session, sending notification" << dendl;
 	    notif->add_watcher(oi_iter->first, Watch::WATCHER_NOTIFIED); // adding before send_message to avoid race
+            entity_name_t name = oi_iter->first;
+            session->add_notif(notif, name);
 
 	    MWatchNotify *notify_msg = new MWatchNotify(w.cookie, w.ver, notif->id, WATCH_NOTIFY);
 	    osd->client_messenger->send_message(notify_msg, session->con);
@@ -1145,7 +1147,7 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops,
           obc->ref++;
           notif->obc = obc;
 	  notif->timeout = new Watch::C_NotifyTimeout(osd, notif);
-	  osd->watch_timer.add_event_after(5.0, notif->timeout);
+	  osd->watch_timer.add_event_after(5.0, notif->timeout); /* FIXME: use a configurable timeout here */
 	}
 	osd->watch_lock.Unlock();
       }
@@ -1176,18 +1178,11 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops,
 	  result = -EINVAL;
 	  break;
 	}
+	OSD::Session *session = (OSD::Session *)ctx->op->get_connection()->get_priv();
+        session->del_notif(notif);
 
 	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;
-	  MWatchNotify *reply = notif->reply; // new MWatchNotify(notif->cookie, wi.ver, notif->id, WATCH_NOTIFY_COMPLETE);
-	  osd->client_messenger->send_message(reply, notif->session->con);
-	  notif->session->put();
-	  osd->watch->remove_notification(notif);
-	  if (notif->timeout)
-	    osd->watch_timer.cancel_event(notif->timeout);
-	  delete notif;
-	}
+        osd->ack_notification(source, notif);
 	osd->watch_lock.Unlock();
       }
       break;
-- 
2.39.5