Merge remote branch 'gh/wip-mon'

16 files changed, 45 insertions, 83 deletions

diff --git a/src/mon/AuthMonitor.cc b/src/mon/AuthMonitor.cc
index 1dd8b7314aa..dac30353c87 100644
--- a/src/mon/AuthMonitor.cc
+++ b/src/mon/AuthMonitor.cc
@@ -260,11 +260,6 @@ bool AuthMonitor::prepare_update(PaxosServiceMessage *m)
   }
 }
 
-void AuthMonitor::committed()
-{
-
-}
-
 void AuthMonitor::election_finished()
 {
   dout(10) << "AuthMonitor::election_starting" << dendl;
diff --git a/src/mon/AuthMonitor.h b/src/mon/AuthMonitor.h
index d60de0fa356..056484540b4 100644
--- a/src/mon/AuthMonitor.h
+++ b/src/mon/AuthMonitor.h
@@ -101,8 +101,6 @@ private:
   uint64_t assign_global_id(MAuth *m, bool should_increase_max);
   void encode_pending(bufferlist &bl);  // propose pending update to peers
 
-  void committed();
-
   bool preprocess_query(PaxosServiceMessage *m);  // true if processed.
   bool prepare_update(PaxosServiceMessage *m);
 
diff --git a/src/mon/LogMonitor.cc b/src/mon/LogMonitor.cc
index 1e305d34242..2122aff3626 100644
--- a/src/mon/LogMonitor.cc
+++ b/src/mon/LogMonitor.cc
@@ -226,11 +226,6 @@ bool LogMonitor::prepare_update(PaxosServiceMessage *m)
   }
 }
 
-void LogMonitor::committed()
-{
-
-}
-
 bool LogMonitor::preprocess_log(MLog *m)
 {
   dout(10) << "preprocess_log " << *m << " from " << m->get_orig_source() << dendl;
diff --git a/src/mon/LogMonitor.h b/src/mon/LogMonitor.h
index 20367c08672..da4c8d4ceda 100644
--- a/src/mon/LogMonitor.h
+++ b/src/mon/LogMonitor.h
@@ -38,8 +38,6 @@ private:
   void create_pending();  // prepare a new pending
   void encode_pending(bufferlist &bl);  // propose pending update to peers
 
-  void committed();
-
   bool preprocess_query(PaxosServiceMessage *m);  // true if processed.
   bool prepare_update(PaxosServiceMessage *m);
 
diff --git a/src/mon/MDSMonitor.cc b/src/mon/MDSMonitor.cc
index 28fa3950eae..57043d0f9e8 100644
--- a/src/mon/MDSMonitor.cc
+++ b/src/mon/MDSMonitor.cc
@@ -472,8 +472,7 @@ void MDSMonitor::_updated(MMDSBeacon *m)
   m->put();
 }
 
-
-void MDSMonitor::committed()
+void MDSMonitor::on_active()
 {
   tick();
 }
diff --git a/src/mon/MDSMonitor.h b/src/mon/MDSMonitor.h
index 6ee80790c52..a315231803c 100644
--- a/src/mon/MDSMonitor.h
+++ b/src/mon/MDSMonitor.h
@@ -77,7 +77,7 @@ class MDSMonitor : public PaxosService {
   bool prepare_update(PaxosServiceMessage *m);
   bool should_propose(double& delay);
 
-  void committed();
+  void on_active();
 
   void _note_beacon(class MMDSBeacon *m);
   bool preprocess_beacon(class MMDSBeacon *m);
diff --git a/src/mon/MonmapMonitor.cc b/src/mon/MonmapMonitor.cc
index f50766c369c..f5d25e6201c 100644
--- a/src/mon/MonmapMonitor.cc
+++ b/src/mon/MonmapMonitor.cc
@@ -367,11 +367,6 @@ bool MonmapMonitor::should_propose(double& delay)
   return true;
 }
 
-void MonmapMonitor::committed()
-{
-  //Nothing useful to do here.
-}
-
 void MonmapMonitor::tick()
 {
   update_from_paxos();
diff --git a/src/mon/MonmapMonitor.h b/src/mon/MonmapMonitor.h
index 0df458cae41..0393de9dff0 100644
--- a/src/mon/MonmapMonitor.h
+++ b/src/mon/MonmapMonitor.h
@@ -69,8 +69,6 @@ class MonmapMonitor : public PaxosService {
    */
   bool should_propose(double& delay);
 
-  void committed();
-
   void tick();
 
  private:
diff --git a/src/mon/OSDMonitor.cc b/src/mon/OSDMonitor.cc
index eef90a2b85c..b29473fc7d8 100644
--- a/src/mon/OSDMonitor.cc
+++ b/src/mon/OSDMonitor.cc
@@ -142,7 +142,9 @@ bool OSDMonitor::update_from_paxos()
 
   send_to_waiting();
   check_subs();
-   
+
+  share_map_with_random_osd();
+
   return true;
 }
 
@@ -294,17 +296,14 @@ void OSDMonitor::encode_pending(bufferlist &bl)
 }
 
 
-void OSDMonitor::committed()
+void OSDMonitor::share_map_with_random_osd()
 {
   // tell any osd
-  int r = osdmap.get_any_up_osd();
-  if (r >= 0) {
-    MonSession *s = mon->session_map.get_random_osd_session();
-    if (s) {
-      dout(10) << "committed, telling random " << s->inst << " all about it" << dendl;
-      MOSDMap *m = build_incremental(osdmap.get_epoch() - 1, osdmap.get_epoch());  // whatev, they'll request more if they need it
-      mon->messenger->send_message(m, s->inst);
-    }
+  MonSession *s = mon->session_map.get_random_osd_session();
+  if (s) {
+    dout(10) << "committed, telling random " << s->inst << " all about it" << dendl;
+    MOSDMap *m = build_incremental(osdmap.get_epoch() - 1, osdmap.get_epoch());  // whatev, they'll request more if they need it
+    mon->messenger->send_message(m, s->inst);
   }
 }
 
diff --git a/src/mon/OSDMonitor.h b/src/mon/OSDMonitor.h
index 6ba34e1ec86..294214885ea 100644
--- a/src/mon/OSDMonitor.h
+++ b/src/mon/OSDMonitor.h
@@ -57,7 +57,7 @@ private:
   void create_pending();  // prepare a new pending
   void encode_pending(bufferlist &bl);
 
-  void committed();
+  void share_map_with_random_osd();
 
   void handle_query(PaxosServiceMessage *m);
   bool preprocess_query(PaxosServiceMessage *m);  // true if processed.
diff --git a/src/mon/PGMonitor.cc b/src/mon/PGMonitor.cc
index 1a21a291a04..8f61c7c36e8 100644
--- a/src/mon/PGMonitor.cc
+++ b/src/mon/PGMonitor.cc
@@ -258,11 +258,6 @@ bool PGMonitor::prepare_update(PaxosServiceMessage *m)
   }
 }
 
-void PGMonitor::committed()
-{
-
-}
-
 void PGMonitor::handle_statfs(MStatfs *statfs)
 {
   // check caps
diff --git a/src/mon/PGMonitor.h b/src/mon/PGMonitor.h
index a5a4d63e20e..a9514feff34 100644
--- a/src/mon/PGMonitor.h
+++ b/src/mon/PGMonitor.h
@@ -52,8 +52,6 @@ private:
   void create_pending();  // prepare a new pending
   void encode_pending(bufferlist &bl);  // propose pending update to peers
 
-  void committed();
-
   bool preprocess_query(PaxosServiceMessage *m);  // true if processed.
   bool prepare_update(PaxosServiceMessage *m);
 
diff --git a/src/mon/Paxos.cc b/src/mon/Paxos.cc
index 437cc393d8b..40600d1b8ba 100644
--- a/src/mon/Paxos.cc
+++ b/src/mon/Paxos.cc
@@ -63,6 +63,8 @@ void Paxos::collect(version_t oldpn)
   uncommitted_v = 0;
   uncommitted_pn = 0;
   uncommitted_value.clear();
+  peer_first_committed.clear();
+  peer_last_committed.clear();
 
   // look for uncommitted value
   if (mon->store->exists_bl_sn(machine_name, last_committed+1)) {
@@ -247,14 +249,9 @@ void Paxos::handle_last(MMonPaxos *last)
     return;
   }
 
-  // share committed values?
-  if (last->last_committed < last_committed) {
-    // share committed values
-    dout(10) << "sending commit to " << last->get_source() << dendl;
-    MMonPaxos *commit = new MMonPaxos(mon->get_epoch(), MMonPaxos::OP_COMMIT, machine_id);
-    share_state(commit, last->first_committed, last->last_committed);
-    mon->messenger->send_message(commit, last->get_source_inst());
-  }
+  // note peer's last_committed, in case we learn a new commit and need to
+  // push it to them.
+  peer_last_committed[last->get_source().num()] = last->last_committed;
 
   // did we receive a committed value?
   store_state(last);
@@ -293,6 +290,21 @@ void Paxos::handle_last(MMonPaxos *last)
       mon->timer.cancel_event(collect_timeout_event);
       collect_timeout_event = 0;
 
+      // share committed values?
+      for (map<int,version_t>::iterator p = peer_last_committed.begin();
+	   p != peer_last_committed.end();
+	   ++p) {
+	if (p->second < last_committed) {
+	  // share committed values
+	  dout(10) << " sending commit to mon." << p->first << dendl;
+	  MMonPaxos *commit = new MMonPaxos(mon->get_epoch(), MMonPaxos::OP_COMMIT, machine_id);
+	  share_state(commit, peer_first_committed[p->first], p->second);
+	  mon->messenger->send_message(commit, mon->monmap->get_inst(p->first));
+	}
+      }
+      peer_first_committed.clear();
+      peer_last_committed.clear();
+
       // almost...
       state = STATE_ACTIVE;
 
diff --git a/src/mon/Paxos.h b/src/mon/Paxos.h
index a340cdab3b9..e67a975f4fa 100644
--- a/src/mon/Paxos.h
+++ b/src/mon/Paxos.h
@@ -112,6 +112,7 @@ private:
   utime_t last_commit_time;
   version_t accepted_pn;
   version_t accepted_pn_from;
+  map<int,version_t> peer_first_committed, peer_last_committed;
 
   // active (phase 2)
   utime_t lease_expire;
diff --git a/src/mon/PaxosService.cc b/src/mon/PaxosService.cc
index 783af679736..399fd071977 100644
--- a/src/mon/PaxosService.cc
+++ b/src/mon/PaxosService.cc
@@ -104,23 +104,6 @@ bool PaxosService::should_propose(double& delay)
   return true;
 }
 
-void PaxosService::_commit()
-{
-  dout(7) << "_commit" << dendl;
-  update_from_paxos();   // notify service of new paxos state
-
-  if (mon->is_leader() && paxos->is_active()) {
-    dout(7) << "_commit creating new pending" << dendl;
-    if (!have_pending) {
-      create_pending();
-      have_pending = true;
-    }
-
-    committed();
-  }
-}
-
-
 void PaxosService::propose_pending()
 {
   dout(10) << "propose_pending" << dendl;
@@ -138,7 +121,7 @@ void PaxosService::propose_pending()
   have_pending = false;
 
   // apply to paxos
-  paxos->wait_for_commit_front(new C_Commit(this));
+  paxos->wait_for_commit_front(new C_Active(this));
   paxos->propose_new_value(bl);
 }
 
@@ -173,8 +156,12 @@ void PaxosService::election_finished()
 
 void PaxosService::_active()
 {
+  if (!paxos->is_active()) {
+    dout(10) << "_active - not active" << dendl;
+    paxos->wait_for_active(new C_Active(this));
+    return;
+  }
   dout(10) << "_active" << dendl;
-  assert(paxos->is_active());
 
   // pull latest from paxos
   update_from_paxos();
@@ -195,7 +182,10 @@ void PaxosService::_active()
     }
   }
 
-  on_active();
+  // NOTE: it's possible that this will get called twice if we commit
+  // an old paxos value.  Implementations should be mindful of that.
+  if (paxos->is_active())
+    on_active();
 }
 
 
diff --git a/src/mon/PaxosService.h b/src/mon/PaxosService.h
index a0cd66ee6b3..c8dc93bd7cf 100644
--- a/src/mon/PaxosService.h
+++ b/src/mon/PaxosService.h
@@ -45,16 +45,6 @@ protected:
 	svc->_active();
     }
   };
-  class C_Commit : public Context {
-    PaxosService *svc;
-  public:
-    C_Commit(PaxosService *s) : svc(s) {}
-    void finish(int r) {
-      if (r >= 0)
-	svc->_commit();
-    }
-  };
-  friend class C_Update;
 
   class C_Propose : public Context {
     PaxosService *ps;
@@ -87,7 +77,6 @@ public:
 
 private:
   void _active();
-  void _commit();
 
 public:
   // i implement and you use
@@ -156,6 +145,8 @@ public:
    * This is called when the Paxos state goes to active.
    * It's a courtesy method if you have things you want/need
    * to do at that time.
+   *
+   * Note that is may get called twice in certain recovery cases.
    */
   virtual void on_active() { }
 
@@ -165,8 +156,6 @@ public:
    */
   virtual void on_restart() { }
 
-  virtual void committed() = 0;            // [leader] called after a proposed value commits
-
   virtual void tick() {}
 
   virtual enum health_status_t get_health(std::ostream& os) const { return HEALTH_OK; }