PG: block outgoing peering messages until the last interval is durable

Signed-off-by: Samuel Just <sam.just@inktank.com>

diff --git a/src/osd/PG.cc b/src/osd/PG.cc
index aa84ae87e8f6eb1b2e395578fd20f13154c51656..173fa7c4117cae1def58cf5c9291c4fa99cf679d 100644 (file)
--- a/src/osd/PG.cc
+++ b/src/osd/PG.cc
@@ -4627,6 +4627,21 @@ void PG::start_flush(ObjectStore::Transaction *t,
   on_safe->push_back(new ContainerContext<FlushStateRef>(flush_trigger));
 }
 
+void PG::reset_interval_flush()
+{
+  dout(10) << "Clearing blocked outgoing recovery messages" << dendl;
+  recovery_state.clear_blocked_outgoing();
+  
+  if (!osr->flush_commit(
+      new QueuePeeringEvt<IntervalFlush>(
+       this, get_osdmap()->get_epoch(), IntervalFlush()))) {
+    dout(10) << "Beginning to block outgoing recovery messages" << dendl;
+    recovery_state.begin_block_outgoing();
+  } else {
+    dout(10) << "Not blocking outgoing recovery messages" << dendl;
+  }
+}
+
 /* Called before initializing peering during advance_map */
 void PG::start_peering_interval(
   const OSDMapRef lastmap,
@@ -4637,6 +4652,7 @@ void PG::start_peering_interval(
   const OSDMapRef osdmap = get_osdmap();
 
   set_last_peering_reset();
+  reset_interval_flush();
 
   vector<int> oldacting, oldup;
   int oldrole = get_role();
@@ -5374,6 +5390,15 @@ PG::RecoveryState::Started::Started(my_context ctx)
   context< RecoveryMachine >().log_enter(state_name);
 }
 
+boost::statechart::result
+PG::RecoveryState::Started::react(const IntervalFlush&)
+{
+  dout(10) << "Ending blocked outgoing recovery messages" << dendl;
+  context< RecoveryMachine >().pg->recovery_state.end_block_outgoing();
+  return discard_event();
+}
+
+
 boost::statechart::result
 PG::RecoveryState::Started::react(const FlushedEvt&)
 {
@@ -5425,6 +5450,7 @@ PG::RecoveryState::Reset::Reset(my_context ctx)
 {
   context< RecoveryMachine >().log_enter(state_name);
   PG *pg = context< RecoveryMachine >().pg;
+
   pg->flushes_in_progress = 0;
   pg->set_last_peering_reset();
 }
@@ -5437,6 +5463,14 @@ PG::RecoveryState::Reset::react(const FlushedEvt&)
   return discard_event();
 }
 
+boost::statechart::result
+PG::RecoveryState::Reset::react(const IntervalFlush&)
+{
+  dout(10) << "Ending blocked outgoing recovery messages" << dendl;
+  context< RecoveryMachine >().pg->recovery_state.end_block_outgoing();
+  return discard_event();
+}
+
 boost::statechart::result PG::RecoveryState::Reset::react(const AdvMap& advmap)
 {
   PG *pg = context< RecoveryMachine >().pg;

diff --git a/src/osd/PG.h b/src/osd/PG.h
index a380db42406ba8b2384a4a8992b92b773ffcd19d..7bf3376ab3cfb9aa7d9037f43e0e5589111ec45e 100644 (file)
--- a/src/osd/PG.h
+++ b/src/osd/PG.h
@@ -1381,6 +1381,8 @@ public:
 
   TrivialEvent(AllReplicasActivated)
 
+  TrivialEvent(IntervalFlush)
+
   /* Encapsulates PG recovery process */
   class RecoveryState {
     void start_handle(RecoveryCtx *new_ctx);
@@ -1499,12 +1501,14 @@ public:
        boost::statechart::custom_reaction< ActMap >,
        boost::statechart::custom_reaction< NullEvt >,
        boost::statechart::custom_reaction< FlushedEvt >,
+       boost::statechart::custom_reaction< IntervalFlush >,
        boost::statechart::transition< boost::statechart::event_base, Crashed >
        > reactions;
       boost::statechart::result react(const QueryState& q);
       boost::statechart::result react(const AdvMap&);
       boost::statechart::result react(const ActMap&);
       boost::statechart::result react(const FlushedEvt&);
+      boost::statechart::result react(const IntervalFlush&);
       boost::statechart::result react(const boost::statechart::event_base&) {
        return discard_event();
       }
@@ -1521,11 +1525,13 @@ public:
        boost::statechart::custom_reaction< AdvMap >,
        boost::statechart::custom_reaction< NullEvt >,
        boost::statechart::custom_reaction< FlushedEvt >,
+       boost::statechart::custom_reaction< IntervalFlush >,
        boost::statechart::transition< boost::statechart::event_base, Crashed >
        > reactions;
       boost::statechart::result react(const QueryState& q);
       boost::statechart::result react(const AdvMap&);
       boost::statechart::result react(const FlushedEvt&);
+      boost::statechart::result react(const IntervalFlush&);
       boost::statechart::result react(const boost::statechart::event_base&) {
        return discard_event();
       }
@@ -2099,6 +2105,7 @@ public:
   /// share new pg log entries after a pg is active
   void share_pg_log();
 
+  void reset_interval_flush();
   void start_peering_interval(
     const OSDMapRef lastmap,
     const vector<int>& newup, int up_primary,