monc: hunt for new monitor on failure; cleanup interface

diff --git a/src/cfuse.cc b/src/cfuse.cc
index 514bd3b0a6267860187b8b1d832aa9aed6a2becd..b8db144241a99fe7297bdb095cc7c9594c251efe 100644 (file)
--- a/src/cfuse.cc
+++ b/src/cfuse.cc
@@ -76,7 +76,7 @@ int main(int argc, const char **argv, const char *envp[]) {
 
   rank.start();
 
-  rank.set_policy(entity_name_t::TYPE_MON, SimpleMessenger::Policy::lossless());
+  rank.set_policy(entity_name_t::TYPE_MON, SimpleMessenger::Policy::lossy_fail_after(1.0));
   rank.set_policy(entity_name_t::TYPE_MDS, SimpleMessenger::Policy::lossless());
   rank.set_policy(entity_name_t::TYPE_OSD, SimpleMessenger::Policy::lossless());
 

diff --git a/src/client/Client.cc b/src/client/Client.cc
index 9bec9fe82bba354e9d93e363432f10b0cb6931dc..d214c8b4db3827926b438b608edd2569fdbbc326 100644 (file)
--- a/src/client/Client.cc
+++ b/src/client/Client.cc
@@ -2760,8 +2760,6 @@ void Client::tick()
   tick_event = new C_C_Tick(this);
   timer.add_event_after(g_conf.client_tick_interval, tick_event);
 
-  monclient->tick();
-  
   utime_t now = g_clock.now();
 
   if (mdsmap->get_epoch()) {

diff --git a/src/csyn.cc b/src/csyn.cc
index 58d61e33df76d748cb7d413de46b943e407c759c..88b87b7dd8737c865a76d800f3c6698494191259 100644 (file)
--- a/src/csyn.cc
+++ b/src/csyn.cc
@@ -59,7 +59,7 @@ int main(int argc, const char **argv, char *envp[])
   rank.bind();
   cout << "starting csyn at " << rank.get_rank_addr() << std::endl;
 
-  rank.set_policy(entity_name_t::TYPE_MON, SimpleMessenger::Policy::lossless());
+  rank.set_policy(entity_name_t::TYPE_MON, SimpleMessenger::Policy::lossy_fail_after(2.0));
   rank.set_policy(entity_name_t::TYPE_MDS, SimpleMessenger::Policy::lossless());
   rank.set_policy(entity_name_t::TYPE_OSD, SimpleMessenger::Policy::lossless());
 

diff --git a/src/mds/MDS.cc b/src/mds/MDS.cc
index 2dc9e1c2c785eb7de60c0c128dcd475c31d33774..cc27e39241f583721860e82a9e978797d484d327 100644 (file)
--- a/src/mds/MDS.cc
+++ b/src/mds/MDS.cc
@@ -497,7 +497,9 @@ void MDS::beacon_send()
   beacon->set_standby_for_rank(standby_for_rank);
   beacon->set_standby_for_name(standby_for_name);
 
-  monc->send_mon_message(beacon, newmon);
+  if (newmon)
+    monc->pick_new_mon();
+  monc->send_mon_message(beacon);
 
   // schedule next sender
   if (beacon_sender) timer.cancel_event(beacon_sender);

diff --git a/src/mon/MonClient.cc b/src/mon/MonClient.cc
index 67dbd6d61b6dd92e1737d1353da28dd3289bfb4b..0d7d609f0627d034732c606b9e9a4dc3b34c9f76 100644 (file)
--- a/src/mon/MonClient.cc
+++ b/src/mon/MonClient.cc
@@ -191,25 +191,11 @@ void MonClient::handle_monmap(MMonMap *m)
 // -------------------
 // MOUNT
 
-void MonClient::_try_mount(double timeout)
+void MonClient::_send_mount()
 {
-  dout(10) << "_try_mount" << dendl;
-  int mon = monmap.pick_mon();
-  dout(2) << "sending client_mount to mon" << mon << dendl;
-  messenger->set_dispatcher(this);
-  messenger->send_message(new MClientMount, monmap.get_inst(mon));
-
-  // schedule timeout?
-  assert(mount_timeout_event == 0);
-  mount_timeout_event = new C_MountTimeout(this, timeout);
-  timer.add_event_after(timeout, mount_timeout_event);
-}
-
-void MonClient::_mount_timeout(double timeout)
-{
-  dout(10) << "_mount_timeout" << dendl;
-  mount_timeout_event = 0;
-  _try_mount(timeout);
+  dout(10) << "_send_mount" << dendl;
+  _send_mon_message(new MClientMount);
+  mount_started = g_clock.now();
 }
 
 int MonClient::mount(double mount_timeout)
@@ -222,38 +208,41 @@ int MonClient::mount(double mount_timeout)
   }
 
   // only first mounter does the work
-  bool itsme = false;
   if (!mounters) {
-    itsme = true;
-    _try_mount(mount_timeout);
-  } else {
+    // init
+    messenger->set_dispatcher(this);
+    timer.add_event_after(10.0, new C_Tick(this));
+  
+    _send_mount();
+  } else
     dout(5) << "additional mounter" << dendl;
-  }
   mounters++;
 
-  while (!mounted)
+  while (!mounted && !mount_err)
     mount_cond.Wait(monc_lock);
 
-  dout(5) << "mount success" << dendl;
-  return 0;
+  if (!mounted) {
+    mounters--;
+  } else {
+    dout(5) << "mount success" << dendl;
+  }
+
+  return mount_err;
 }
 
 void MonClient::handle_mount_ack(MClientMountAck* m)
 {
   dout(10) << "handle_mount_ack " << *m << dendl;
 
+  hunting = false;
+
   // monmap
   bufferlist::iterator p = m->monmap_bl.begin();
   ::decode(monmap, p);
 
   messenger->set_myname(entity_name_t::CLIENT(m->client.v));
 
-  // finish.
-  timer.cancel_event(mount_timeout_event);
-  mount_timeout_event = 0;
-
   mounted = true;
-
   mount_cond.SignalAll();
   
   delete m;
@@ -262,33 +251,63 @@ void MonClient::handle_mount_ack(MClientMountAck* m)
 
 // ---------
 
-void MonClient::send_mon_message(Message *m, bool newmon)
+void MonClient::_send_mon_message(Message *m)
 {
-  Mutex::Locker l(monc_lock);
-  int mon = monmap.pick_mon(newmon);
+  int mon = monmap.pick_mon();
   messenger->send_message(m, monmap.mon_inst[mon]);
 }
 
-void MonClient::pick_new_mon()
+void MonClient::_pick_new_mon()
 {
-  Mutex::Locker l(monc_lock);
   int oldmon = monmap.pick_mon();
   messenger->mark_down(monmap.get_inst(oldmon).addr);
   monmap.pick_mon(true);
 }
 
 
-void MonClient::ms_handle_remote_reset(const entity_addr_t& peer)
+void MonClient::ms_handle_reset(const entity_addr_t& peer)
 {
-  dout(10) << "ms_handle_peer_reset " << peer << dendl;
-  pick_new_mon();
-  renew_subs();
+  dout(10) << "ms_handle_reset " << peer << dendl;
+  if (!hunting) {
+    dout(0) << "staring hunt for new mon" << dendl;
+    hunting = true;
+    _pick_new_mon();
+    if (!mounted)
+      _send_mount();
+    _renew_subs();
+  }
+}
+
+void MonClient::tick()
+{
+  dout(10) << "tick" << dendl;
+
+  if (hunting) {
+    dout(0) << "continuing hunt mounted=" << mounted << dendl;
+    // try new monitor
+    _pick_new_mon();
+    if (!mounted)
+      _send_mount();
+    _renew_subs();
+  } else {
+    // just renew as needed
+    utime_t now = g_clock.now();
+    if (mounted && now > sub_renew_after)
+      _renew_subs();
+
+    int oldmon = monmap.pick_mon();
+    messenger->send_keepalive(monmap.mon_inst[oldmon]);
+  }
+
+  timer.add_event_after(10.0, new C_Tick(this));
+  dout(10) << "tick done" << dendl;
+
 }
 
 
 // ---------
 
-void MonClient::renew_subs()
+void MonClient::_renew_subs()
 {
   if (sub_have.empty()) {
     dout(10) << "renew_subs - empty" << dendl;
@@ -302,11 +321,13 @@ void MonClient::renew_subs()
 
   MMonSubscribe *m = new MMonSubscribe;
   m->what = sub_have;
-  send_mon_message(m);
+  _send_mon_message(m);
 }
 
 void MonClient::handle_subscribe_ack(MMonSubscribeAck *m)
 {
+  hunting = false;
+
   if (sub_renew_sent != utime_t()) {
     sub_renew_after = sub_renew_sent;
     sub_renew_after += m->interval / 2.0;
@@ -317,24 +338,3 @@ void MonClient::handle_subscribe_ack(MMonSubscribeAck *m)
   }
 }
 
-void MonClient::tick()
-{
-  if (!mounted)
-    return;
-
-  utime_t now = g_clock.now();
-  static utime_t last_tick;
-
-  if (now - last_tick < 10.0)
-    return;
-  last_tick = now;
-
-  dout(10) << "tick" << dendl;
-
-  if (now > sub_renew_after)
-    renew_subs();
-
-  int oldmon = monmap.pick_mon();
-  messenger->send_keepalive(monmap.mon_inst[oldmon]);
-}
-

diff --git a/src/mon/MonClient.h b/src/mon/MonClient.h
index 7e960c5b2485f0e6bf344698ce1d940426ac0f89..93c99dfae3b48acaf4668d5125a7e9ac05523080 100644 (file)
--- a/src/mon/MonClient.h
+++ b/src/mon/MonClient.h
@@ -38,53 +38,72 @@ private:
 
   entity_addr_t my_addr;
 
-  Context *mount_timeout_event;
-
   Mutex monc_lock;
   SafeTimer timer;
-  bool mounted;
-  int mounters;
-  Cond mount_cond, map_cond;
 
   bool ms_dispatch(Message *m);
   void handle_monmap(MMonMap *m);
 
-  void ms_handle_remote_reset(const entity_addr_t& peer);
+  void ms_handle_reset(const entity_addr_t& peer);
+
 
- protected:
-  class C_MountTimeout : public Context {
-    MonClient *client;
-    double timeout;
-  public:
-    C_MountTimeout(MonClient *c, double to) : client(c), timeout(to) { }
+  // monitor session
+  bool hunting;
+
+  struct C_Tick : public Context {
+    MonClient *monc;
+    C_Tick(MonClient *m) : monc(m) {}
     void finish(int r) {
-      if (r >= 0) client->_mount_timeout(timeout);
+      monc->tick();
     }
   };
+  void tick();
 
-  void _try_mount(double timeout);
-  void _mount_timeout(double timeout);
+  // mount
+private:
+  bool mounted;
+  int mount_err;
+  int mounters;
+  Cond mount_cond, map_cond;
+  utime_t mount_started;
+
+  void _pick_new_mon();
+  void _send_mon_message(Message *m);
+  void _send_mount();
   void handle_mount_ack(MClientMountAck* m);
 
+public:
+  int mount(double mount_timeout);
+
+
   // mon subscriptions
 private:
   map<nstring,ceph_mon_subscribe_item> sub_have;  // my subs, and current versions
   utime_t sub_renew_sent, sub_renew_after;
 
+  void _renew_subs();
+  void handle_subscribe_ack(MMonSubscribeAck* m);
+
 public:
-  void renew_subs();
+  void renew_subs() {
+    Mutex::Locker l(monc_lock);
+    _renew_subs();
+  }
   void sub_want(nstring what, version_t have) {
+    Mutex::Locker l(monc_lock);
     sub_have[what].have = have;
     sub_have[what].onetime = false;
   }
   void sub_want_onetime(nstring what, version_t have) {
+    Mutex::Locker l(monc_lock);
     if (sub_have.count(what) == 0) {
       sub_have[what].have = have;
       sub_have[what].onetime = true;
-      renew_subs();
+      _renew_subs();
     }
   }
   void sub_got(nstring what, version_t have) {
+    Mutex::Locker l(monc_lock);
     if (sub_have.count(what)) {
       if (sub_have[what].onetime)
        sub_have.erase(what);
@@ -92,29 +111,35 @@ public:
        sub_have[what].have = have;
     }
   }
-  void handle_subscribe_ack(MMonSubscribeAck* m);
 
  public:
   MonClient() : messenger(NULL),
-               mount_timeout_event(NULL),
                monc_lock("MonClient::monc_lock"),
                timer(monc_lock) {
+    hunting = false;
     mounted = false;
     mounters = 0;
+    mount_err = 0;
+  }
+  ~MonClient() {
+    timer.cancel_all_events();
   }
 
   int build_initial_monmap();
   int get_monmap();
 
-  int mount(double mount_timeout);
-
-  void tick();
-
-  void send_mon_message(Message *m, bool new_mon=false);
+  void send_mon_message(Message *m) {
+    Mutex::Locker l(monc_lock);
+    _send_mon_message(m);
+  }
   void note_mon_leader(int m) {
+    Mutex::Locker l(monc_lock);
     monmap.last_mon = m;
   }
-  void pick_new_mon();
+  void pick_new_mon() {
+    Mutex::Locker l(monc_lock);
+    _pick_new_mon();
+  }
 
   entity_addr_t get_my_addr() { return my_addr; }