From: Sage Weil Date: Fri, 4 Sep 2009 18:08:03 +0000 (-0700) Subject: monc: hunt for new monitor on failure; cleanup interface X-Git-Tag: v0.14~30 X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=72ccd2037a1750ef8f3a96d4ae6ba51906b22422;p=ceph.git monc: hunt for new monitor on failure; cleanup interface --- diff --git a/src/cfuse.cc b/src/cfuse.cc index 514bd3b0a62..b8db144241a 100644 --- 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 9bec9fe82bb..d214c8b4db3 100644 --- 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 58d61e33df7..88b87b7dd87 100644 --- 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 2dc9e1c2c78..cc27e39241f 100644 --- 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 67dbd6d61b6..0d7d609f062 100644 --- 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 7e960c5b248..93c99dfae3b 100644 --- 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 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; }