osd: helpers to blacklist messages to down osds

There is a race between handle_osd_map -> note_down_osd() and PG threads:

 - handle_osd_map -> note_down_osd marks down an osd for epoch N
 - a pg thread with epoch <N sends a message to the (old) peer, reopening
   the msgr connection
 - nobody cleans up

Introduce a pre_publish_map() OSDService method and helpers for sending
messages to peers.  Pass in the epoch we are working from, and drop the
message on the floor if the target OSD has been since marked down.

See #3548.

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

diff --git a/src/osd/OSD.cc b/src/osd/OSD.cc
index 96052820a15729c78c1b2384ca9908d1d3ca81c9..914d1dcc88ab8ca03c58fa62fb5f77c6b5e0827f 100644 (file)
--- a/src/osd/OSD.cc
+++ b/src/osd/OSD.cc
@@ -2475,6 +2475,49 @@ void OSD::send_alive()
   }
 }
 
+void OSDService::send_message_osd_cluster(int peer, Message *m, epoch_t from_epoch)
+{
+  Mutex::Locker l(publish_lock);
+
+  // service map is always newer/newest
+  assert(from_epoch <= next_osdmap->get_epoch());
+
+  if (next_osdmap->is_down(peer) ||
+      next_osdmap->get_info(peer).up_from > from_epoch) {
+    m->put();
+    return;
+  }
+  osd->cluster_messenger->send_message(m, next_osdmap->get_cluster_inst(peer));
+}
+
+Connection *OSDService::get_con_osd_cluster(int peer, epoch_t from_epoch)
+{
+  Mutex::Locker l(publish_lock);
+
+  // service map is always newer/newest
+  assert(from_epoch <= next_osdmap->get_epoch());
+
+  if (next_osdmap->is_down(peer) ||
+      next_osdmap->get_info(peer).up_from > from_epoch) {
+    return NULL;
+  }
+  return osd->cluster_messenger->get_connection(next_osdmap->get_cluster_inst(peer));
+}
+
+Connection *OSDService::get_con_osd_hb(int peer, epoch_t from_epoch)
+{
+  Mutex::Locker l(publish_lock);
+
+  // service map is always newer/newest
+  assert(from_epoch <= next_osdmap->get_epoch());
+
+  if (next_osdmap->is_down(peer) ||
+      next_osdmap->get_info(peer).up_from > from_epoch) {
+    return NULL;
+  }
+  return osd->hbclient_messenger->get_connection(next_osdmap->get_hb_inst(peer));
+}
+
 void OSDService::queue_want_pg_temp(pg_t pgid, vector<int>& want)
 {
   Mutex::Locker l(pg_temp_lock);
@@ -3632,14 +3675,19 @@ void OSD::handle_osd_map(MOSDMap *m)
     OSDMapRef newmap = get_map(cur);
     assert(newmap);  // we just cached it above!
 
+    // start blacklisting messages sent to peers that go down.
+    service.pre_publish_map(newmap);
+
     // kill connections to newly down osds
     set<int> old;
     osdmap->get_all_osds(old);
-    for (set<int>::iterator p = old.begin(); p != old.end(); p++)
+    for (set<int>::iterator p = old.begin(); p != old.end(); p++) {
       if (*p != whoami &&
          osdmap->have_inst(*p) &&                        // in old map
-         (!newmap->exists(*p) || !newmap->is_up(*p)))    // but not the new one
+         (!newmap->exists(*p) || !newmap->is_up(*p))) {  // but not the new one
        note_down_osd(*p);
+      }
+    }
     
     osdmap = newmap;
 

diff --git a/src/osd/OSD.h b/src/osd/OSD.h
index ff277e1e5b8bce48af094e62affa4abdcd6dbae9..4608891a20ddda04787eca76b57c3b4d859087f8 100644 (file)
--- a/src/osd/OSD.h
+++ b/src/osd/OSD.h
@@ -193,18 +193,28 @@ public:
     Mutex::Locker l(publish_lock);
     superblock = block;
   }
-  OSDMapRef osdmap;
+  OSDMapRef osdmap, next_osdmap;
   OSDMapRef get_osdmap() {
     Mutex::Locker l(publish_lock);
     return osdmap;
   }
+  void pre_publish_map(OSDMapRef map) {
+    Mutex::Locker l(publish_lock);
+    next_osdmap = map;
+  }
   void publish_map(OSDMapRef map) {
     Mutex::Locker l(publish_lock);
     osdmap = map;
+    next_osdmap = map;
   }
 
   int get_nodeid() const { return whoami; }
 
+  // -- message helpers --
+  Connection *get_con_osd_cluster(int peer, epoch_t from_epoch);
+  Connection *get_con_osd_hb(int peer, epoch_t from_epoch);
+  void send_message_osd_cluster(int peer, Message *m, epoch_t from_epoch);
+
   // -- scrub scheduling --
   Mutex sched_scrub_lock;
   int scrubs_pending;